From bf4acdfb8588aeba37864eb7a81d7d463acbef6b Mon Sep 17 00:00:00 2001 From: fengzero Date: Sat, 7 Oct 2023 08:47:17 +0000 Subject: [PATCH] feat(core): trans_data msg are sent using a separate channel --- include/neuron/adapter.h | 181 +++++++++++-- plugins/ekuiper/json_rw.c | 146 ++--------- plugins/ekuiper/plugin_ekuiper.c | 7 - plugins/monitor/monitor.c | 2 - plugins/mqtt/mqtt_handle.c | 130 ++-------- plugins/mqtt/mqtt_plugin.c | 10 - plugins/restful/rest.c | 2 - plugins/restful/rw_handle.c | 112 +------- src/adapter/adapter.c | 164 ++++++++++-- src/adapter/adapter_internal.h | 14 +- src/adapter/driver/driver.c | 371 ++++++++++++++++++--------- src/adapter/driver/driver_internal.h | 5 + src/core/manager.c | 106 ++------ src/core/manager.h | 2 + src/core/manager_internal.c | 28 +- src/core/manager_internal.h | 11 +- src/core/storage.c | 11 +- src/parser/neu_json_group_config.h | 7 +- 18 files changed, 666 insertions(+), 643 deletions(-) diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index d28237b3..704939e2 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -24,8 +24,11 @@ extern "C" { #endif +#include #include +#include "json/neu_json_rw.h" + #include "define.h" #include "metrics.h" #include "tag.h" @@ -38,7 +41,6 @@ typedef struct { typedef enum neu_reqresp_type { NEU_RESP_ERROR, - NEU_REQ_UPDATE_LICENSE, NEU_REQ_READ_GROUP, NEU_RESP_READ_GROUP, @@ -145,8 +147,7 @@ typedef enum neu_reqresp_type { } neu_reqresp_type_e; static const char *neu_reqresp_type_string_t[] = { - [NEU_RESP_ERROR] = "NEU_RESP_ERROR", - [NEU_REQ_UPDATE_LICENSE] = "NEU_REQ_UPDATE_LICENSE", + [NEU_RESP_ERROR] = "NEU_RESP_ERROR", [NEU_REQ_READ_GROUP] = "NEU_REQ_READ_GROUP", [NEU_RESP_READ_GROUP] = "NEU_RESP_READ_GROUP", @@ -475,10 +476,11 @@ typedef struct neu_resp_get_tag { } neu_resp_get_tag_t; typedef struct { - char app[NEU_NODE_NAME_LEN]; - char driver[NEU_NODE_NAME_LEN]; - char group[NEU_GROUP_NAME_LEN]; - char *params; + char app[NEU_NODE_NAME_LEN]; + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; + uint16_t port; + char * params; } neu_req_subscribe_t; typedef struct { @@ -488,9 +490,10 @@ typedef struct { } neu_req_unsubscribe_t; typedef struct { - char *driver; - char *group; - char *params; + char * driver; + char * group; + uint16_t port; + char * params; } neu_req_subscribe_group_info_t; typedef struct { @@ -768,9 +771,6 @@ static inline void neu_req_inst_templates_fini(neu_req_inst_templates_t *req) free(req->insts); } -typedef struct neu_req_update_license { -} neu_req_update_license_t; - typedef struct neu_req_read_group { char driver[NEU_NODE_NAME_LEN]; char group[NEU_GROUP_NAME_LEN]; @@ -794,6 +794,12 @@ typedef struct neu_resp_tag_value_meta { neu_tag_meta_t metas[NEU_TAG_META_SIZE]; } neu_resp_tag_value_meta_t; +static inline UT_icd *neu_resp_tag_value_meta_icd() +{ + static UT_icd icd = { sizeof(neu_resp_tag_value_meta_t), NULL, NULL, NULL }; + return &icd; +} + typedef struct neu_req_write_tags { char driver[NEU_NODE_NAME_LEN]; char group[NEU_GROUP_NAME_LEN]; @@ -803,19 +809,150 @@ typedef struct neu_req_write_tags { } neu_req_write_tags_t; typedef struct { - char driver[NEU_NODE_NAME_LEN]; - char group[NEU_GROUP_NAME_LEN]; - uint16_t n_tag; - neu_resp_tag_value_meta_t *tags; + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; + + UT_array *tags; // neu_resp_tag_value_meta_t } neu_resp_read_group_t; +static inline void neu_resp_read_free(neu_resp_read_group_t *resp) +{ + utarray_foreach(resp->tags, neu_resp_tag_value_meta_t *, tag_value) + { + if (tag_value->value.type == NEU_TYPE_PTR) { + free(tag_value->value.value.ptr.ptr); + } + } + utarray_free(resp->tags); +} + typedef struct { - char driver[NEU_NODE_NAME_LEN]; - char group[NEU_GROUP_NAME_LEN]; - uint16_t n_tag; - neu_resp_tag_value_meta_t tags[]; + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; + + uint16_t index; + pthread_mutex_t mtx; + UT_array * tags; // neu_resp_tag_value_meta_t } neu_reqresp_trans_data_t; +static inline void neu_trans_data_free(neu_reqresp_trans_data_t *data) +{ + pthread_mutex_lock(&data->mtx); + if (data->index > 0) { + data->index -= 1; + } + + if (data->index == 0) { + utarray_foreach(data->tags, neu_resp_tag_value_meta_t *, tag_value) + { + if (tag_value->value.type == NEU_TYPE_PTR) { + free(tag_value->value.value.ptr.ptr); + } + } + utarray_free(data->tags); + pthread_mutex_unlock(&data->mtx); + pthread_mutex_destroy(&data->mtx); + } else { + pthread_mutex_unlock(&data->mtx); + } +} + +static inline void neu_tag_value_to_json(neu_resp_tag_value_meta_t *tag_value, + neu_json_read_resp_tag_t * tag_json) +{ + tag_json->name = tag_value->tag; + tag_json->error = 0; + + for (int k = 0; k < NEU_TAG_META_SIZE; k++) { + if (strlen(tag_value->metas[k].name) > 0) { + tag_json->n_meta++; + } else { + break; + } + } + if (tag_json->n_meta > 0) { + tag_json->metas = (neu_json_tag_meta_t *) calloc( + tag_json->n_meta, sizeof(neu_json_tag_meta_t)); + } + neu_json_metas_to_json(tag_value->metas, NEU_TAG_META_SIZE, tag_json); + + switch (tag_value->value.type) { + case NEU_TYPE_ERROR: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.i32; + tag_json->error = tag_value->value.value.i32; + break; + case NEU_TYPE_UINT8: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.u8; + break; + case NEU_TYPE_INT8: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.i8; + break; + case NEU_TYPE_INT16: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.i16; + break; + case NEU_TYPE_INT32: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.i32; + break; + case NEU_TYPE_INT64: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.i64; + break; + case NEU_TYPE_WORD: + case NEU_TYPE_UINT16: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.u16; + break; + case NEU_TYPE_DWORD: + case NEU_TYPE_UINT32: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.u32; + break; + case NEU_TYPE_LWORD: + case NEU_TYPE_UINT64: + tag_json->t = NEU_JSON_INT; + tag_json->value.val_int = tag_value->value.value.u64; + break; + case NEU_TYPE_FLOAT: + tag_json->t = NEU_JSON_FLOAT; + tag_json->value.val_float = tag_value->value.value.f32; + tag_json->precision = tag_value->value.precision; + break; + case NEU_TYPE_DOUBLE: + tag_json->t = NEU_JSON_DOUBLE; + tag_json->value.val_double = tag_value->value.value.d64; + tag_json->precision = tag_value->value.precision; + break; + case NEU_TYPE_BOOL: + tag_json->t = NEU_JSON_BOOL; + tag_json->value.val_bool = tag_value->value.value.boolean; + break; + case NEU_TYPE_BIT: + tag_json->t = NEU_JSON_BIT; + tag_json->value.val_bit = tag_value->value.value.u8; + break; + case NEU_TYPE_STRING: + tag_json->t = NEU_JSON_STR; + tag_json->value.val_str = tag_value->value.value.str; + break; + case NEU_TYPE_PTR: + tag_json->t = NEU_JSON_STR; + tag_json->value.val_str = (char *) tag_value->value.value.ptr.ptr; + break; + case NEU_TYPE_BYTES: + tag_json->t = NEU_JSON_BYTES; + tag_json->value.val_bytes.length = tag_value->value.value.bytes.length; + tag_json->value.val_bytes.bytes = tag_value->value.value.bytes.bytes; + break; + default: + break; + } +} + typedef struct { char node[NEU_NODE_NAME_LEN]; } neu_reqresp_node_deleted_t; @@ -943,6 +1080,8 @@ typedef struct adapter_callbacks { int (*command)(neu_adapter_t *adapter, neu_reqresp_head_t head, void *data); int (*response)(neu_adapter_t *adapter, neu_reqresp_head_t *head, void *data); + int (*responseto)(neu_adapter_t *adapter, neu_reqresp_head_t *head, + void *data, struct sockaddr_in dst); neu_adapter_register_metric_cb_t register_metric; neu_adapter_update_metric_cb_t update_metric; diff --git a/plugins/ekuiper/json_rw.c b/plugins/ekuiper/json_rw.c index 42df0813..0ce06a4b 100644 --- a/plugins/ekuiper/json_rw.c +++ b/plugins/ekuiper/json_rw.c @@ -25,111 +25,6 @@ #include "json_rw.h" #include "plugin_ekuiper.h" -int wrap_tag_data(neu_json_read_resp_tag_t * json_tag, - neu_resp_tag_value_meta_t *tag_value) -{ - if (NULL == json_tag || NULL == tag_value) { - return -1; - } - - for (int k = 0; k < NEU_TAG_META_SIZE; k++) { - if (strlen(tag_value->metas[k].name) > 0) { - json_tag->n_meta++; - } else { - break; - } - } - - if (json_tag->n_meta > 0) { - json_tag->metas = (neu_json_tag_meta_t *) calloc( - json_tag->n_meta, sizeof(neu_json_tag_meta_t)); - if (json_tag->metas == NULL) { - return -1; - } - } - - neu_json_metas_to_json(tag_value->metas, NEU_TAG_META_SIZE, json_tag); - - json_tag->name = tag_value->tag; - json_tag->error = NEU_ERR_SUCCESS; - - switch (tag_value->value.type) { - case NEU_TYPE_INT8: - json_tag->t = NEU_JSON_INT; - json_tag->value.val_int = tag_value->value.value.i8; - break; - case NEU_TYPE_UINT8: - json_tag->t = NEU_JSON_INT; - json_tag->value.val_int = tag_value->value.value.u8; - break; - case NEU_TYPE_INT16: - json_tag->t = NEU_JSON_INT; - json_tag->value.val_int = tag_value->value.value.i16; - break; - case NEU_TYPE_WORD: - case NEU_TYPE_UINT16: - json_tag->t = NEU_JSON_INT; - json_tag->value.val_int = tag_value->value.value.u16; - break; - case NEU_TYPE_INT32: - json_tag->t = NEU_JSON_INT; - json_tag->value.val_int = tag_value->value.value.i32; - break; - case NEU_TYPE_DWORD: - case NEU_TYPE_UINT32: - json_tag->t = NEU_JSON_INT; - json_tag->value.val_int = tag_value->value.value.u32; - break; - case NEU_TYPE_INT64: - json_tag->t = NEU_JSON_INT; - json_tag->value.val_int = tag_value->value.value.i64; - break; - case NEU_TYPE_LWORD: - case NEU_TYPE_UINT64: - json_tag->t = NEU_JSON_INT; - json_tag->value.val_int = tag_value->value.value.u64; - break; - case NEU_TYPE_FLOAT: - json_tag->t = NEU_JSON_FLOAT; - json_tag->value.val_float = tag_value->value.value.f32; - break; - case NEU_TYPE_DOUBLE: - json_tag->t = NEU_JSON_DOUBLE; - json_tag->value.val_double = tag_value->value.value.d64; - break; - case NEU_TYPE_BIT: - json_tag->t = NEU_JSON_BIT; - json_tag->value.val_bit = tag_value->value.value.u8; - break; - case NEU_TYPE_BOOL: - json_tag->t = NEU_JSON_BOOL; - json_tag->value.val_bool = tag_value->value.value.boolean; - break; - case NEU_TYPE_STRING: - 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; - json_tag->error = tag_value->value.value.i32; - break; - case NEU_TYPE_BYTES: - json_tag->t = NEU_JSON_BYTES; - json_tag->value.val_bytes.length = tag_value->value.value.bytes.length; - json_tag->value.val_bytes.bytes = tag_value->value.value.bytes.bytes; - break; - default: - break; - } - - return 0; -} - int json_encode_read_resp_header(void *json_object, void *param) { int ret = 0; @@ -186,18 +81,17 @@ int json_encode_read_resp_tags(void *json_object, void *param) return -1; } - for (int i = 0; i < trans_data->n_tag; i++) { + utarray_foreach(trans_data->tags, neu_resp_tag_value_meta_t *, tag_value) + { neu_json_read_resp_tag_t json_tag = { 0 }; - if (0 != wrap_tag_data(&json_tag, &trans_data->tags[i])) { - continue; // ignore - } + neu_tag_value_to_json(tag_value, &json_tag); neu_json_elem_t tag_elem = { .name = json_tag.name, .t = json_tag.t, .v = json_tag.value, - .precision = trans_data->tags[i].value.precision, + .precision = tag_value->value.precision, }; if (json_tag.n_meta > 0) { @@ -231,22 +125,24 @@ int json_encode_read_resp_tags(void *json_object, void *param) } } - neu_json_elem_t resp_elems[] = { { - .name = "values", - .t = NEU_JSON_OBJECT, - .v.val_object = values_object, - }, - { - .name = "errors", - .t = NEU_JSON_OBJECT, - .v.val_object = errors_object, + neu_json_elem_t resp_elems[] = { + { + .name = "values", + .t = NEU_JSON_OBJECT, + .v.val_object = values_object, + }, + { + .name = "errors", + .t = NEU_JSON_OBJECT, + .v.val_object = errors_object, - }, - { - .name = "metas", - .t = NEU_JSON_OBJECT, - .v.val_object = metas_object, - } }; + }, + { + .name = "metas", + .t = NEU_JSON_OBJECT, + .v.val_object = metas_object, + }, + }; // steals `values_object` and `errors_object` ret = neu_json_encode_field(json_object, resp_elems, NEU_JSON_ELEM_SIZE(resp_elems)); diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 9c19e5f6..a36e3962 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -308,11 +308,6 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin, } 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); - } - } break; } case NEU_REQ_UPDATE_NODE: { @@ -329,8 +324,6 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin, } case NEU_REQ_UNSUBSCRIBE_GROUP: break; - case NEU_REQ_UPDATE_LICENSE: - break; default: plog_warn(plugin, "unsupported request type: %d", header->type); break; diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index f396a34f..a4f1d639 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -332,8 +332,6 @@ static int monitor_plugin_request(neu_plugin_t * plugin, handle_events(plugin, header->type, data); break; } - case NEU_REQ_UPDATE_LICENSE: - break; default: nlog_warn("recv unsupported msg: %s", neu_reqresp_type_string(header->type)); diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 5b4b1875..f1738ae5 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -27,117 +27,25 @@ #include "mqtt_handle.h" #include "mqtt_plugin.h" -static int tag_values_to_json(neu_resp_tag_value_meta_t *tags, uint16_t len, - neu_json_read_resp_t *json) +static int tag_values_to_json(UT_array *tags, neu_json_read_resp_t *json) { - if (0 == len) { + int index = 0; + + if (0 == utarray_len(tags)) { return 0; } - json->n_tag = len; + json->n_tag = utarray_len(tags); json->tags = (neu_json_read_resp_tag_t *) calloc( json->n_tag, sizeof(neu_json_read_resp_tag_t)); if (NULL == json->tags) { return -1; } - for (int i = 0; i < json->n_tag; i++) { - neu_resp_tag_value_meta_t *tag = &tags[i]; - neu_type_e type = tag->value.type; - json->tags[i].name = tag->tag; - json->tags[i].error = NEU_ERR_SUCCESS; - - for (int k = 0; k < NEU_TAG_META_SIZE; k++) { - if (strlen(tag->metas[k].name) > 0) { - json->tags[i].n_meta++; - } else { - break; - } - } - - if (json->tags[i].n_meta > 0) { - json->tags[i].metas = (neu_json_tag_meta_t *) calloc( - json->tags[i].n_meta, sizeof(neu_json_tag_meta_t)); - } - - neu_json_metas_to_json(tag->metas, NEU_TAG_META_SIZE, &json->tags[i]); - - switch (type) { - case NEU_TYPE_ERROR: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.i32; - json->tags[i].error = tag->value.value.i32; - break; - case NEU_TYPE_UINT8: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.u8; - break; - case NEU_TYPE_INT8: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.i8; - break; - case NEU_TYPE_INT16: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.i16; - break; - case NEU_TYPE_INT32: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.i32; - break; - case NEU_TYPE_INT64: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.i64; - break; - case NEU_TYPE_WORD: - case NEU_TYPE_UINT16: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.u16; - break; - case NEU_TYPE_DWORD: - case NEU_TYPE_UINT32: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.u32; - break; - case NEU_TYPE_LWORD: - case NEU_TYPE_UINT64: - json->tags[i].t = NEU_JSON_INT; - json->tags[i].value.val_int = tag->value.value.u64; - break; - case NEU_TYPE_FLOAT: - json->tags[i].t = NEU_JSON_FLOAT; - json->tags[i].value.val_float = tag->value.value.f32; - json->tags[i].precision = tag->value.precision; - break; - case NEU_TYPE_DOUBLE: - json->tags[i].t = NEU_JSON_DOUBLE; - json->tags[i].value.val_double = tag->value.value.d64; - json->tags[i].precision = tag->value.precision; - break; - case NEU_TYPE_BOOL: - json->tags[i].t = NEU_JSON_BOOL; - json->tags[i].value.val_bool = tag->value.value.boolean; - break; - case NEU_TYPE_BIT: - json->tags[i].t = NEU_JSON_BIT; - json->tags[i].value.val_bit = tag->value.value.u8; - break; - case NEU_TYPE_STRING: - 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; - case NEU_TYPE_BYTES: - json->tags[i].t = NEU_JSON_BYTES; - json->tags[i].value.val_bytes.length = - tag->value.value.bytes.length; - json->tags[i].value.val_bytes.bytes = tag->value.value.bytes.bytes; - break; - default: - break; - } + utarray_foreach(tags, neu_resp_tag_value_meta_t *, tag_value) + { + neu_tag_value_to_json(tag_value, &json->tags[index]); + index += 1; } return 0; @@ -147,15 +55,13 @@ static char *generate_upload_json(neu_plugin_t * plugin, neu_reqresp_trans_data_t *data, mqtt_upload_format_e format) { - neu_resp_tag_value_meta_t *tags = data->tags; - uint16_t len = data->n_tag; - char * json_str = NULL; - neu_json_read_periodic_t header = { .group = (char *) data->group, + char * json_str = NULL; + neu_json_read_periodic_t header = { .group = (char *) data->group, .node = (char *) data->driver, .timestamp = global_timestamp }; - neu_json_read_resp_t json = { 0 }; + neu_json_read_resp_t json = { 0 }; - if (0 != tag_values_to_json(tags, len, &json)) { + if (0 != tag_values_to_json(data->tags, &json)) { plog_error(plugin, "tag_values_to_json fail"); return NULL; } @@ -188,12 +94,12 @@ static char *generate_read_resp_json(neu_plugin_t * plugin, neu_json_mqtt_t * mqtt, neu_resp_read_group_t *data) { - neu_resp_tag_value_meta_t *tags = data->tags; - uint16_t len = data->n_tag; - char * json_str = NULL; - neu_json_read_resp_t json = { 0 }; + // neu_resp_tag_value_meta_t *tags = data->tags; + // uint16_t len = data->n_tag; + char * json_str = NULL; + neu_json_read_resp_t json = { 0 }; - if (0 != tag_values_to_json(tags, len, &json)) { + if (0 != tag_values_to_json(data->tags, &json)) { plog_error(plugin, "tag_values_to_json fail"); return NULL; } diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 890c8e87..b5c1a9d7 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -354,14 +354,6 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, break; 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: @@ -382,8 +374,6 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, case NEU_REQRESP_NODE_DELETED: error = handle_del_driver(plugin, data); break; - case NEU_REQ_UPDATE_LICENSE: - break; default: error = NEU_ERR_MQTT_FAILURE; break; diff --git a/plugins/restful/rest.c b/plugins/restful/rest.c index 94c8b76f..17e1faf7 100644 --- a/plugins/restful/rest.c +++ b/plugins/restful/rest.c @@ -240,8 +240,6 @@ static int dashb_plugin_request(neu_plugin_t * plugin, handle_get_ndriver_tags_resp(header->ctx, (neu_resp_get_ndriver_tags_t *) data); break; - case NEU_REQ_UPDATE_LICENSE: - break; default: nlog_fatal("recv unhandle msg: %s", neu_reqresp_type_string(header->type)); diff --git a/plugins/restful/rw_handle.c b/plugins/restful/rw_handle.c index ea686dbf..9d548430 100644 --- a/plugins/restful/rw_handle.c +++ b/plugins/restful/rw_handle.c @@ -207,116 +207,19 @@ void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp) { neu_json_read_resp_t api_res = { 0 }; char * result = NULL; + int index = 0; - api_res.n_tag = resp->n_tag; + api_res.n_tag = utarray_len(resp->tags); api_res.tags = calloc(api_res.n_tag, sizeof(neu_json_read_resp_tag_t)); - for (int i = 0; i < resp->n_tag; i++) { - neu_resp_tag_value_meta_t *tag = &resp->tags[i]; - - api_res.tags[i].name = resp->tags[i].tag; - api_res.tags[i].error = NEU_ERR_SUCCESS; - - for (int k = 0; k < NEU_TAG_META_SIZE; k++) { - if (strlen(tag->metas[k].name) > 0) { - api_res.tags[i].n_meta++; - } else { - break; - } - } - - if (api_res.tags[i].n_meta > 0) { - api_res.tags[i].metas = (neu_json_tag_meta_t *) calloc( - api_res.tags[i].n_meta, sizeof(neu_json_tag_meta_t)); - } - - neu_json_metas_to_json(tag->metas, NEU_TAG_META_SIZE, &api_res.tags[i]); - - switch (resp->tags[i].value.type) { - case NEU_TYPE_INT8: - api_res.tags[i].t = NEU_JSON_INT; - api_res.tags[i].value.val_int = resp->tags[i].value.value.i8; - break; - case NEU_TYPE_UINT8: - api_res.tags[i].t = NEU_JSON_INT; - api_res.tags[i].value.val_int = resp->tags[i].value.value.u8; - break; - case NEU_TYPE_INT16: - api_res.tags[i].t = NEU_JSON_INT; - api_res.tags[i].value.val_int = resp->tags[i].value.value.i16; - break; - case NEU_TYPE_WORD: - case NEU_TYPE_UINT16: - api_res.tags[i].t = NEU_JSON_INT; - api_res.tags[i].value.val_int = resp->tags[i].value.value.u16; - break; - case NEU_TYPE_INT32: - api_res.tags[i].t = NEU_JSON_INT; - api_res.tags[i].value.val_int = resp->tags[i].value.value.i32; - break; - case NEU_TYPE_DWORD: - case NEU_TYPE_UINT32: - api_res.tags[i].t = NEU_JSON_INT; - api_res.tags[i].value.val_int = resp->tags[i].value.value.u32; - break; - case NEU_TYPE_INT64: - api_res.tags[i].t = NEU_JSON_INT; - api_res.tags[i].value.val_int = resp->tags[i].value.value.i64; - break; - case NEU_TYPE_LWORD: - case NEU_TYPE_UINT64: - api_res.tags[i].t = NEU_JSON_INT; - api_res.tags[i].value.val_int = resp->tags[i].value.value.u64; - break; - case NEU_TYPE_FLOAT: - api_res.tags[i].t = NEU_JSON_FLOAT; - api_res.tags[i].value.val_float = resp->tags[i].value.value.f32; - api_res.tags[i].precision = resp->tags[i].value.precision; - break; - case NEU_TYPE_DOUBLE: - api_res.tags[i].t = NEU_JSON_DOUBLE; - api_res.tags[i].value.val_double = resp->tags[i].value.value.d64; - api_res.tags[i].precision = resp->tags[i].value.precision; - break; - case NEU_TYPE_BIT: - api_res.tags[i].t = NEU_JSON_BIT; - api_res.tags[i].value.val_bit = resp->tags[i].value.value.u8; - break; - case NEU_TYPE_BOOL: - api_res.tags[i].t = NEU_JSON_BOOL; - api_res.tags[i].value.val_bool = resp->tags[i].value.value.boolean; - break; - case NEU_TYPE_STRING: - api_res.tags[i].t = NEU_JSON_STR; - api_res.tags[i].value.val_str = resp->tags[i].value.value.str; - break; - case NEU_TYPE_BYTES: - api_res.tags[i].t = NEU_JSON_BYTES; - api_res.tags[i].value.val_bytes.length = - resp->tags[i].value.value.bytes.length; - api_res.tags[i].value.val_bytes.bytes = - resp->tags[i].value.value.bytes.bytes; - break; - case NEU_TYPE_ERROR: - api_res.tags[i].t = NEU_JSON_INT; - 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; - } + utarray_foreach(resp->tags, neu_resp_tag_value_meta_t *, tag_value) + { + neu_tag_value_to_json(tag_value, &api_res.tags[index]); + index += 1; } 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); - } + for (int i = 0; i < api_res.n_tag; i++) { if (api_res.tags[i].n_meta > 0) { free(api_res.tags[i].metas); } @@ -324,5 +227,4 @@ void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp) neu_http_ok(aio, result); free(api_res.tags); free(result); - free(resp->tags); } diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 5eba82ec..f43cf5c1 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -38,11 +38,16 @@ #include "plugin.h" #include "storage.h" +static int adapter_trans_data(enum neu_event_io_type type, int fd, + void *usr_data); static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data); static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, void *data); static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data); +static int adapter_responseto(neu_adapter_t * adapter, + neu_reqresp_head_t *header, void *data, + struct sockaddr_in dst); static int adapter_register_metric(neu_adapter_t *adapter, const char *name, const char *help, neu_metric_type_e type, uint64_t init); @@ -57,6 +62,7 @@ inline static void notify_monitor(neu_adapter_t * adapter, static const adapter_callbacks_t callback_funs = { .command = adapter_command, .response = adapter_response, + .responseto = adapter_responseto, .register_metric = adapter_register_metric, .update_metric = adapter_update_metric, }; @@ -118,6 +124,14 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load) adapter->control_fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP); if (adapter->control_fd <= 0) { + free(adapter); + return NULL; + } + adapter->trans_data_fd = + socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP); + if (adapter->trans_data_fd <= 0) { + close(adapter->control_fd); + free(adapter); return NULL; } @@ -127,10 +141,12 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load) adapter->handle = info->handle; adapter->cb_funs.command = callback_funs.command; adapter->cb_funs.response = callback_funs.response; + adapter->cb_funs.responseto = callback_funs.responseto; adapter->cb_funs.register_metric = callback_funs.register_metric; adapter->cb_funs.update_metric = callback_funs.update_metric; adapter->module = info->module; adapter->timestamp_lev = 0; + adapter->trans_data_port = 0; struct sockaddr_in remote = { .sin_family = AF_INET, @@ -149,12 +165,34 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load) neu_adapter_driver_init((neu_adapter_driver_t *) adapter); break; case NEU_NA_TYPE_NDRIVER: - case NEU_NA_TYPE_APP: + case NEU_NA_TYPE_APP: { + while (true) { + uint16_t port = neu_manager_get_port(); + struct sockaddr_in local = { + .sin_family = AF_INET, + .sin_port = htons(port), + .sin_addr.s_addr = inet_addr("127.0.0.1"), + }; + if (bind(adapter->trans_data_fd, (struct sockaddr *) &local, + sizeof(struct sockaddr_in)) == 0) { + adapter->trans_data_port = port; + break; + } + } + + param.usr_data = (void *) adapter; + param.cb = adapter_trans_data; + param.fd = adapter->trans_data_fd; + + adapter->trans_data_io = neu_event_add_io(adapter->events, param); + if (adapter->module->display) { REGISTER_APP_METRICS(adapter); } + break; } + } adapter->plugin = adapter->module->intf_funs->open(); assert(adapter->plugin != NULL); @@ -198,6 +236,8 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load) if (adapter->module->type == NEU_NA_TYPE_DRIVER) { neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter); + } else { + neu_event_del_io(adapter->events, adapter->trans_data_io); } neu_event_del_io(adapter->events, adapter->control_io); @@ -219,6 +259,11 @@ bool neu_adapter_reset(neu_adapter_t *adapter, neu_adapter_info_t *info) return true; } +uint16_t neu_adapter_trans_data_port(neu_adapter_t *adapter) +{ + return adapter->trans_data_port; +} + int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name) { char *name = strdup(new_name); @@ -461,22 +506,12 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, void *data) { - if (header->type == NEU_REQRESP_TRANS_DATA) { - return 0; - } - - switch (header->type) { - case NEU_REQRESP_TRANS_DATA: - strcpy(header->sender, adapter->name); - break; - default: - neu_msg_exchange(header); - break; - } + assert(header->type != NEU_REQRESP_TRANS_DATA); + neu_msg_exchange(header); neu_msg_gen(header, data); int ret = send(adapter->control_fd, header, header->len, 0); - if (ret != 0) { + if (ret <= 0) { nlog_error("adapter: %s send response %s failed, ret: %d, errno: %d", adapter->name, neu_reqresp_type_string(header->type), ret, errno); @@ -485,6 +520,65 @@ static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, return ret; } +static int adapter_responseto(neu_adapter_t * adapter, + neu_reqresp_head_t *header, void *data, + struct sockaddr_in dst) +{ + assert(header->type == NEU_REQRESP_TRANS_DATA); + strcpy(header->sender, adapter->name); + + neu_msg_gen(header, data); + int ret = sendto(adapter->control_fd, header, header->len, 0, + (struct sockaddr *) &dst, sizeof(dst)); + if (ret <= 0) { + nlog_error("adapter: %s send responseto %s failed, ret: %d, errno: %d", + adapter->name, neu_reqresp_type_string(header->type), ret, + errno); + } + + return ret; +} + +static int adapter_trans_data(enum neu_event_io_type type, int fd, + void *usr_data) +{ + neu_adapter_t *adapter = (neu_adapter_t *) usr_data; + if (type != NEU_EVENT_IO_READ) { + nlog_warn("adapter: %s recv close, exit loop, fd: %d", adapter->name, + fd); + return 0; + } + + memset(adapter->recv_buf, 0, sizeof(adapter->recv_buf)); + neu_reqresp_head_t *header = (neu_reqresp_head_t *) adapter->recv_buf; + + int rv = recv(adapter->trans_data_fd, adapter->recv_buf, + sizeof(adapter->recv_buf), 0); + if (rv <= 0) { + nlog_warn("adapter: %s recv trans data failed, ret: %d, errno: %s(%d)", + adapter->name, rv, strerror(errno), errno); + return 0; + } + + nlog_debug("adapter(%s) recv msg from: %s %p, type: %s", adapter->name, + header->sender, header->ctx, + neu_reqresp_type_string(header->type)); + + if (header->type != NEU_REQRESP_TRANS_DATA && + header->type != NEU_RESP_ERROR) { + nlog_warn("adapter: %s recv msg type error, type: %s", adapter->name, + neu_reqresp_type_string(header->type)); + return 0; + } + + adapter->module->intf_funs->request( + adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); + if (header->type == NEU_REQRESP_TRANS_DATA) { + neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]); + } + return 0; +} + static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) { neu_adapter_t *adapter = (neu_adapter_t *) usr_data; @@ -510,19 +604,36 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_reqresp_type_string(header->type)); switch (header->type) { + case NEU_REQ_SUBSCRIBE_GROUP: { + neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1]; + if (adapter->module->type == NEU_NA_TYPE_DRIVER) { + neu_adapter_driver_subscribe((neu_adapter_driver_t *) adapter, cmd); + } else { + adapter->module->intf_funs->request( + adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); + } + break; + } + case NEU_REQ_UNSUBSCRIBE_GROUP: { + neu_req_unsubscribe_t *cmd = (neu_req_unsubscribe_t *) &header[1]; + if (adapter->module->type == NEU_NA_TYPE_DRIVER) { + neu_adapter_driver_unsubscribe((neu_adapter_driver_t *) adapter, + cmd); + } else { + adapter->module->intf_funs->request( + adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); + } + break; + } + case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: case NEU_RESP_GET_DRIVER_GROUP: case NEU_REQRESP_NODE_DELETED: case NEU_RESP_GET_SUB_DRIVER_TAGS: - case NEU_REQ_UPDATE_LICENSE: case NEU_REQ_UPDATE_NODE: case NEU_RESP_GET_NODE_STATE: case NEU_RESP_GET_NODES_STATE: case NEU_RESP_GET_NODE_SETTING: case NEU_REQ_UPDATE_GROUP: - case NEU_REQ_SUBSCRIBE_GROUP: - case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: - case NEU_REQ_UNSUBSCRIBE_GROUP: - case NEU_RESP_READ_GROUP: case NEU_RESP_GET_SUBSCRIBE_GROUP: case NEU_RESP_ADD_TAG: case NEU_RESP_ADD_TEMPLATE_TAG: @@ -538,7 +649,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_GET_NDRIVER_TAGS: case NEU_RESP_GET_GROUP: case NEU_RESP_ERROR: - case NEU_REQRESP_TRANS_DATA: case NEU_REQRESP_NODES_STATE: case NEU_REQ_ADD_NODE_EVENT: case NEU_REQ_DEL_NODE_EVENT: @@ -553,6 +663,11 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); break; + case NEU_RESP_READ_GROUP: + adapter->module->intf_funs->request( + adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); + neu_resp_read_free((neu_resp_read_group_t *) &header[1]); + break; case NEU_REQ_READ_GROUP: { neu_resp_error_t error = { 0 }; @@ -1033,6 +1148,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } default: + nlog_warn("adapter: %s recv msg type error, type: %s", adapter->name, + neu_reqresp_type_string(header->type)); assert(false); break; } @@ -1044,6 +1161,7 @@ void neu_adapter_destroy(neu_adapter_t *adapter) { nlog_notice("adapter %s destroy", adapter->name); close(adapter->control_fd); + close(adapter->trans_data_fd); adapter->module->intf_funs->close(adapter->plugin); @@ -1385,6 +1503,9 @@ void neu_msg_gen(neu_reqresp_head_t *header, void *data) size_t data_size = 0; switch (header->type) { + case NEU_REQRESP_TRANS_DATA: + data_size = sizeof(neu_reqresp_trans_data_t); + break; case NEU_REQ_NODE_INIT: case NEU_REQ_NODE_UNINIT: case NEU_RESP_NODE_UNINIT: @@ -1588,9 +1709,6 @@ void neu_msg_gen(neu_reqresp_head_t *header, void *data) case NEU_RESP_READ_GROUP: data_size = sizeof(neu_resp_read_group_t); break; - case NEU_REQ_UPDATE_LICENSE: - data_size = sizeof(neu_req_update_license_t); - break; case NEU_REQRESP_NODE_DELETED: data_size = sizeof(neu_reqresp_node_deleted_t); break; diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index b85ea226..2ff2d75d 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -39,10 +39,16 @@ struct neu_adapter { neu_plugin_module_t *module; neu_plugin_t * plugin; - uint8_t buf[NEU_MSG_MAX_SIZE]; - uint8_t recv_buf[NEU_MSG_MAX_SIZE]; + uint8_t buf[NEU_MSG_MAX_SIZE]; + uint8_t recv_buf[NEU_MSG_MAX_SIZE]; + neu_event_io_t *control_io; - int control_fd; + neu_event_io_t *trans_data_io; + + int control_fd; + int trans_data_fd; + + uint16_t trans_data_port; neu_events_t *events; @@ -63,6 +69,8 @@ typedef struct adapter_msg_handler { int neu_adapter_error(); void neu_adapter_set_error(int error); +uint16_t neu_adapter_trans_data_port(neu_adapter_t *adapter); + neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load); bool neu_adapter_reset(neu_adapter_t *adapter, neu_adapter_info_t *info); void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state); diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index da23c315..e18708b3 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -43,6 +43,11 @@ typedef struct to_be_write_tag { void * req; } to_be_write_tag_t; +typedef struct { + char app[NEU_NODE_NAME_LEN]; + struct sockaddr_in addr; +} sub_app_t; + typedef struct group { char *name; @@ -56,6 +61,9 @@ typedef struct group { neu_event_timer_t *read; neu_event_timer_t *write; + UT_array * apps; // sub_app_t array + pthread_mutex_t apps_mtx; + neu_plugin_group_t grp; neu_adapter_driver_t *driver; @@ -78,10 +86,10 @@ static int write_callback(void *usr_data); static void read_group(int64_t timestamp, int64_t timeout, neu_tag_cache_type_e cache_type, neu_driver_cache_t *cache, const char *group, - UT_array *tags, neu_resp_tag_value_meta_t *datas); -static int read_report_group(int64_t timestamp, int64_t timeout, + UT_array *tags, UT_array *tag_values); +static void read_report_group(int64_t timestamp, int64_t timeout, neu_driver_cache_t *cache, const char *group, - UT_array *tags, neu_resp_tag_value_meta_t *datas); + UT_array *tags, UT_array *tag_values); static void update(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value); static void update_im(neu_adapter_t *adapter, const char *group, @@ -161,8 +169,9 @@ static void update_im(neu_adapter_t *adapter, const char *group, const char *tag, neu_dvalue_t value, neu_tag_meta_t *metas, int n_meta) { - neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter; - neu_reqresp_head_t header = { 0 }; + neu_adapter_driver_t * driver = (neu_adapter_driver_t *) adapter; + static __thread uint8_t buf[1024] = { 0 }; + neu_reqresp_head_t * header = (neu_reqresp_head_t *) buf; if (tag == NULL || value.type == NEU_TYPE_ERROR) { nlog_warn("update_im tag is null or value is error %d", @@ -198,20 +207,45 @@ static void update_im(neu_adapter_t *adapter, const char *group, driver->adapter.name, group, tag, neu_type_string(value.type), global_timestamp); - header.type = NEU_REQRESP_TRANS_DATA; - neu_reqresp_trans_data_t *data = calloc( - 1, - sizeof(neu_reqresp_trans_data_t) + sizeof(neu_resp_tag_value_meta_t)); + memset(buf, 0, sizeof(buf)); + header->type = NEU_REQRESP_TRANS_DATA; + neu_reqresp_trans_data_t *data = + calloc(1, sizeof(neu_reqresp_trans_data_t)); strcpy(data->driver, driver->adapter.name); strcpy(data->group, group); + utarray_new(data->tags, neu_resp_tag_value_meta_icd()); - data->n_tag = - read_report_group(global_timestamp, 0, driver->cache, group, tags, - (neu_resp_tag_value_meta_t *) &data[1]); + read_report_group(global_timestamp, 0, driver->cache, group, tags, + data->tags); - if (data->n_tag > 0) { - driver->adapter.cb_funs.response(&driver->adapter, &header, data); + if (utarray_len(data->tags) > 0) { + group_t *find = NULL; + HASH_FIND_STR(driver->groups, group, find); + if (find != NULL) { + pthread_mutex_lock(&find->apps_mtx); + + data->index = utarray_len(find->apps); + if (data->index > 0) { + pthread_mutex_init(&data->mtx, NULL); + + utarray_foreach(find->apps, sub_app_t *, app) + { + if (driver->adapter.cb_funs.responseto( + &driver->adapter, header, data, app->addr) <= 0) { + neu_trans_data_free(data); + } + } + } else { + utarray_free(data->tags); + } + + pthread_mutex_unlock(&find->apps_mtx); + } else { + utarray_free(data->tags); + } + } else { + utarray_free(data->tags); } utarray_free(tags); @@ -289,6 +323,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver) utarray_free(el->static_tags); utarray_free(el->wt_tags); + utarray_free(el->apps); neu_group_destroy(el->group); free(el); } @@ -356,19 +391,18 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, neu_group_t * group = g->group; UT_array * tags = neu_group_get_read_tag(group); - resp.n_tag = utarray_len(tags); - resp.tags = - calloc(1, utarray_len(tags) * sizeof(neu_resp_tag_value_meta_t)); + utarray_new(resp.tags, neu_resp_tag_value_meta_icd()); if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { utarray_foreach(tags, neu_datatag_t *, tag) { - int index = utarray_eltidx(tags, tag); - strcpy(resp.tags[index].tag, tag->name); - resp.tags[index].value.type = NEU_TYPE_ERROR; - resp.tags[index].value.value.i32 = NEU_ERR_PLUGIN_NOT_RUNNING; - } + neu_resp_tag_value_meta_t tag_value = { 0 }; + strcpy(tag_value.tag, tag->name); + tag_value.value.type = NEU_TYPE_ERROR; + tag_value.value.value.i32 = NEU_ERR_PLUGIN_NOT_RUNNING; + utarray_push_back(resp.tags, &tag_value); + } } else { read_group(global_timestamp, neu_group_get_interval(group) * @@ -681,9 +715,10 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name, uint32_t interval) { - UT_icd icd = { sizeof(to_be_write_tag_t), NULL, NULL, NULL }; - group_t *find = NULL; - int ret = NEU_ERR_GROUP_EXIST; + UT_icd icd = { sizeof(to_be_write_tag_t), NULL, NULL, NULL }; + UT_icd sub_icd = { sizeof(sub_app_t), NULL, NULL, NULL }; + group_t *find = NULL; + int ret = NEU_ERR_GROUP_EXIST; HASH_FIND_STR(driver->groups, name, find); if (find == NULL) { @@ -697,8 +732,10 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name, }; pthread_mutex_init(&find->wt_mtx, NULL); + pthread_mutex_init(&find->apps_mtx, NULL); utarray_new(find->wt_tags, &icd); + utarray_new(find->apps, &sub_icd); find->driver = driver; find->name = strdup(name); @@ -881,8 +918,10 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name) utarray_free(find->static_tags); utarray_free(find->grp.tags); utarray_free(find->wt_tags); + utarray_free(find->apps); neu_group_destroy(find->group); pthread_mutex_destroy(&find->wt_mtx); + pthread_mutex_destroy(&find->apps_mtx); free(find); neu_adapter_del_group_metrics(&driver->adapter, name); @@ -1203,34 +1242,55 @@ UT_array *neu_adapter_driver_get_ptag(neu_adapter_driver_t *driver, static int report_callback(void *usr_data) { - group_t * group = (group_t *) usr_data; - neu_reqresp_head_t header = { 0 }; - neu_node_running_state_e state = group->driver->adapter.state; + group_t * group = (group_t *) usr_data; + static __thread uint8_t buf[1024] = { 0 }; + neu_reqresp_head_t * header = (neu_reqresp_head_t *) buf; + neu_node_running_state_e state = group->driver->adapter.state; if (state != NEU_NODE_RUNNING_STATE_RUNNING) { return 0; } - header.type = NEU_REQRESP_TRANS_DATA; + memset(buf, 0, sizeof(buf)); + header->type = NEU_REQRESP_TRANS_DATA; UT_array *tags = neu_adapter_driver_get_read_tag(group->driver, group->name); neu_reqresp_trans_data_t *data = - calloc(1, - sizeof(neu_reqresp_trans_data_t) + - utarray_len(tags) * sizeof(neu_resp_tag_value_meta_t)); + calloc(1, sizeof(neu_reqresp_trans_data_t)); strcpy(data->driver, group->driver->adapter.name); strcpy(data->group, group->name); - data->n_tag = read_report_group(global_timestamp, - neu_group_get_interval(group->group) * - NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, - group->driver->cache, group->name, tags, - (neu_resp_tag_value_meta_t *) &data[1]); + utarray_new(data->tags, neu_resp_tag_value_meta_icd()); - if (data->n_tag > 0) { - group->driver->adapter.cb_funs.response(&group->driver->adapter, - &header, data); + read_report_group(global_timestamp, + neu_group_get_interval(group->group) * + NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, + group->driver->cache, group->name, tags, data->tags); + + if (utarray_len(data->tags) > 0) { + pthread_mutex_lock(&group->apps_mtx); + + data->index = utarray_len(group->apps); + + if (data->index > 0) { + pthread_mutex_init(&data->mtx, NULL); + + utarray_foreach(group->apps, sub_app_t *, app) + { + if (group->driver->adapter.cb_funs.responseto( + &group->driver->adapter, header, data, app->addr) <= + 0) { + neu_trans_data_free(data); + } + } + } else { + utarray_free(data->tags); + } + + pthread_mutex_unlock(&group->apps_mtx); + } else { + utarray_free(data->tags); } utarray_free(tags); free(data); @@ -1368,39 +1428,40 @@ static int read_callback(void *usr_data) return 0; } -static int read_report_group(int64_t timestamp, int64_t timeout, - neu_driver_cache_t *cache, const char *group, - UT_array *tags, neu_resp_tag_value_meta_t *datas) +static void read_report_group(int64_t timestamp, int64_t timeout, + neu_driver_cache_t *cache, const char *group, + UT_array *tags, UT_array *tag_values) { - int index = 0; - utarray_foreach(tags, neu_datatag_t *, tag) { - neu_driver_cache_value_t value = { 0 }; + neu_driver_cache_value_t value = { 0 }; + neu_resp_tag_value_meta_t tag_value = { 0 }; if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) { if (neu_driver_cache_meta_get_changed(cache, group, tag->name, - &value, datas[index].metas, + &value, tag_value.metas, NEU_TAG_META_SIZE) != 0) { nlog_info("tag: %s not changed", tag->name); continue; } } else { if (neu_driver_cache_meta_get(cache, group, tag->name, &value, - datas[index].metas, + tag_value.metas, NEU_TAG_META_SIZE) != 0) { - strcpy(datas[index].tag, tag->name); - datas[index].value.type = NEU_TYPE_ERROR; - datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; - index += 1; + strcpy(tag_value.tag, tag->name); + tag_value.value.type = NEU_TYPE_ERROR; + tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; + + utarray_push_back(tag_values, &tag_value); continue; } } - strcpy(datas[index].tag, tag->name); + strcpy(tag_value.tag, tag->name); if (value.value.type == NEU_TYPE_ERROR) { - datas[index].value = value.value; - index += 1; + tag_value.value = value.value; + + utarray_push_back(tag_values, &tag_value); continue; } @@ -1462,95 +1523,96 @@ static int read_report_group(int64_t timestamp, int64_t timeout, if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) && (timestamp - value.timestamp) > timeout && timeout > 0) { - datas[index].value.type = NEU_TYPE_ERROR; - datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED; + tag_value.value.type = NEU_TYPE_ERROR; + tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED; } else { 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; + tag_value.value.type = NEU_TYPE_PTR; + tag_value.value.value.ptr.length = value.value.value.ptr.length; + tag_value.value.value.ptr.type = value.value.value.ptr.type; + tag_value.value.value.ptr.ptr = value.value.value.ptr.ptr; } else { - datas[index].value = value.value; + tag_value.value = value.value; } if (tag->decimal != 0) { - datas[index].value.type = NEU_TYPE_DOUBLE; + tag_value.value.type = NEU_TYPE_DOUBLE; switch (tag->type) { case NEU_TYPE_INT8: - datas[index].value.value.d64 = - (double) datas[index].value.value.i8 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.i8 * tag->decimal; break; case NEU_TYPE_UINT8: - datas[index].value.value.d64 = - (double) datas[index].value.value.u8 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.u8 * tag->decimal; break; case NEU_TYPE_INT16: - datas[index].value.value.d64 = - (double) datas[index].value.value.i16 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.i16 * tag->decimal; break; case NEU_TYPE_UINT16: - datas[index].value.value.d64 = - (double) datas[index].value.value.u16 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.u16 * tag->decimal; break; case NEU_TYPE_INT32: - datas[index].value.value.d64 = - (double) datas[index].value.value.i32 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.i32 * tag->decimal; break; case NEU_TYPE_UINT32: - datas[index].value.value.d64 = - (double) datas[index].value.value.u32 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.u32 * tag->decimal; break; case NEU_TYPE_INT64: - datas[index].value.value.d64 = - (double) datas[index].value.value.i64 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.i64 * tag->decimal; break; case NEU_TYPE_UINT64: - datas[index].value.value.d64 = - (double) datas[index].value.value.u64 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.u64 * tag->decimal; break; case NEU_TYPE_FLOAT: - datas[index].value.value.d64 = - (double) datas[index].value.value.f32 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.f32 * tag->decimal; break; case NEU_TYPE_DOUBLE: - datas[index].value.value.d64 = - (double) datas[index].value.value.d64 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.d64 * tag->decimal; break; default: - datas[index].value.type = tag->type; + tag_value.value.type = tag->type; break; } } } - index += 1; - } - return index; + utarray_push_back(tag_values, &tag_value); + } } static void read_group(int64_t timestamp, int64_t timeout, neu_tag_cache_type_e cache_type, neu_driver_cache_t *cache, const char *group, - UT_array *tags, neu_resp_tag_value_meta_t *datas) + UT_array *tags, UT_array *tag_values) { utarray_foreach(tags, neu_datatag_t *, tag) { - neu_driver_cache_value_t value = { 0 }; - int index = utarray_eltidx(tags, tag); + neu_resp_tag_value_meta_t tag_value = { 0 }; + neu_driver_cache_value_t value = { 0 }; - strcpy(datas[index].tag, tag->name); + strcpy(tag_value.tag, tag->name); if (neu_driver_cache_meta_get(cache, group, tag->name, &value, - datas[index].metas, + tag_value.metas, NEU_TAG_META_SIZE) != 0) { - datas[index].value.type = NEU_TYPE_ERROR; - datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; + tag_value.value.type = NEU_TYPE_ERROR; + tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; + + utarray_push_back(tag_values, &tag_value); continue; } if (value.value.type == NEU_TYPE_ERROR) { - datas[index].value = value.value; + tag_value.value = value.value; + utarray_push_back(tag_values, &tag_value); continue; } @@ -1613,67 +1675,67 @@ static void read_group(int64_t timestamp, int64_t timeout, if (cache_type != NEU_TAG_CACHE_TYPE_NEVER && !neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) && (timestamp - value.timestamp) > timeout) { - datas[index].value.type = NEU_TYPE_ERROR; - datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED; + tag_value.value.type = NEU_TYPE_ERROR; + tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED; } else { 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; + tag_value.value.type = NEU_TYPE_PTR; + tag_value.value.value.ptr.length = value.value.value.ptr.length; + tag_value.value.value.ptr.type = value.value.value.ptr.type; + tag_value.value.value.ptr.ptr = value.value.value.ptr.ptr; } else { - datas[index].value = value.value; + tag_value.value = value.value; } if (tag->decimal != 0) { - datas[index].value.type = NEU_TYPE_DOUBLE; + tag_value.value.type = NEU_TYPE_DOUBLE; switch (tag->type) { case NEU_TYPE_INT8: - datas[index].value.value.d64 = - (double) datas[index].value.value.i8 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.i8 * tag->decimal; break; case NEU_TYPE_UINT8: - datas[index].value.value.d64 = - (double) datas[index].value.value.u8 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.u8 * tag->decimal; break; case NEU_TYPE_INT16: - datas[index].value.value.d64 = - (double) datas[index].value.value.i16 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.i16 * tag->decimal; break; case NEU_TYPE_UINT16: - datas[index].value.value.d64 = - (double) datas[index].value.value.u16 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.u16 * tag->decimal; break; case NEU_TYPE_INT32: - datas[index].value.value.d64 = - (double) datas[index].value.value.i32 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.i32 * tag->decimal; break; case NEU_TYPE_UINT32: - datas[index].value.value.d64 = - (double) datas[index].value.value.u32 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.u32 * tag->decimal; break; case NEU_TYPE_INT64: - datas[index].value.value.d64 = - (double) datas[index].value.value.i64 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.i64 * tag->decimal; break; case NEU_TYPE_UINT64: - datas[index].value.value.d64 = - (double) datas[index].value.value.u64 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.u64 * tag->decimal; break; case NEU_TYPE_FLOAT: - datas[index].value.value.d64 = - (double) datas[index].value.value.f32 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.f32 * tag->decimal; break; case NEU_TYPE_DOUBLE: - datas[index].value.value.d64 = - (double) datas[index].value.value.d64 * tag->decimal; + tag_value.value.value.d64 = + (double) tag_value.value.value.d64 * tag->decimal; break; default: - datas[index].value.type = tag->type; + tag_value.value.type = tag->type; break; } } } + utarray_push_back(tag_values, &tag_value); } } @@ -1683,3 +1745,60 @@ static void store_write_tag(group_t *group, to_be_write_tag_t *tag) utarray_push_back(group->wt_tags, tag); pthread_mutex_unlock(&group->wt_mtx); } + +void neu_adapter_driver_subscribe(neu_adapter_driver_t *driver, + neu_req_subscribe_t * req) +{ + sub_app_t sub_app = { 0 }; + group_t * find = NULL; + + HASH_FIND_STR(driver->groups, req->group, find); + if (find == NULL) { + nlog_warn("%s sub group: %s not exist", driver->adapter.name, + req->group); + return; + } + + pthread_mutex_lock(&find->apps_mtx); + utarray_foreach(find->apps, sub_app_t *, app) + { + if (strcmp(app->app, req->app) == 0) { + nlog_warn("%s sub group: %s app: %s already exist", + driver->adapter.name, req->group, req->app); + pthread_mutex_unlock(&find->apps_mtx); + return; + } + } + + strcpy(sub_app.app, req->app); + sub_app.addr.sin_family = AF_INET; + sub_app.addr.sin_addr.s_addr = inet_addr("127.0.0.1"); + sub_app.addr.sin_port = htons(req->port); + + utarray_push_back(find->apps, &sub_app); + pthread_mutex_unlock(&find->apps_mtx); +} + +void neu_adapter_driver_unsubscribe(neu_adapter_driver_t * driver, + neu_req_unsubscribe_t *req) +{ + group_t *find = NULL; + + HASH_FIND_STR(driver->groups, req->group, find); + if (find == NULL) { + nlog_warn("%s unsub group: %s not exist", driver->adapter.name, + req->group); + return; + } + + pthread_mutex_lock(&find->apps_mtx); + utarray_foreach(find->apps, sub_app_t *, app) + { + if (strcmp(app->app, req->app) == 0) { + utarray_erase(find->apps, utarray_eltidx(find->apps, app), 1); + pthread_mutex_unlock(&find->apps_mtx); + return; + } + } + pthread_mutex_unlock(&find->apps_mtx); +} \ No newline at end of file diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index 0365e079..c7cfbb2a 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -79,4 +79,9 @@ void neu_adapter_driver_get_value_tag(neu_adapter_driver_t *driver, const char *group, UT_array **tags); UT_array *neu_adapter_driver_get_read_tag(neu_adapter_driver_t *driver, const char * group); + +void neu_adapter_driver_subscribe(neu_adapter_driver_t *driver, + neu_req_subscribe_t * req); +void neu_adapter_driver_unsubscribe(neu_adapter_driver_t * driver, + neu_req_unsubscribe_t *req); #endif diff --git a/src/core/manager.c b/src/core/manager.c index d9f62773..d3f39890 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -50,7 +50,6 @@ #define DEFAULT_DASHBOARD_ADAPTER_NAME DEFAULT_DASHBOARD_PLUGIN_NAME 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); @@ -75,6 +74,11 @@ typedef struct { char *plugin; bool ret; } template_each_cb_data_t; +uint16_t neu_manager_get_port() +{ + static uint16_t port = 10000; + return port++; +} neu_manager_t *neu_manager_create() { @@ -225,37 +229,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) nlog_info("manager recv msg from: %s to %s, type: %s", header->sender, header->receiver, neu_reqresp_type_string(header->type)); switch (header->type) { - case NEU_REQRESP_TRANS_DATA: { - - // neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) - // &header[1]; UT_array *apps = - // neu_subscribe_manager_find(manager->subscribe_manager, cmd->driver, - // cmd->group); - // if (apps != NULL) { - // utarray_foreach(apps, neu_app_subscribe_t *, app) - //{ - // forward_msg_dup(manager, msg, app->pipe, trans_data_dup); - // nlog_debug("forward trans data to pipe: %d", app->pipe.id); - //} - // utarray_free(apps); - //} - - // trans_data_free(msg); - break; - } - case NEU_REQ_UPDATE_LICENSE: { - // UT_array *pipes = - // neu_node_manager_get_pipes_all(manager->node_manager); - - // utarray_foreach(pipes, nng_pipe *, pipe) - //{ - // forward_msg_dup(manager, msg, *pipe, NULL); - // nlog_notice("forward license update to pipe: %d", pipe->id); - //} - // utarray_free(pipes); - - break; - } case NEU_REQ_NODE_INIT: { neu_req_node_init_t *init = (neu_req_node_init_t *) &header[1]; @@ -1043,14 +1016,17 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } case NEU_REQ_SUBSCRIBE_GROUP: { - neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1]; - neu_resp_error_t error = { 0 }; + neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1]; + neu_resp_error_t error = { 0 }; + uint16_t app_port = 0; error.error = neu_manager_subscribe(manager, cmd->app, cmd->driver, - cmd->group, cmd->params); + cmd->group, cmd->params, &app_port); if (error.error == NEU_ERR_SUCCESS) { + cmd->port = app_port; forward_msg(manager, header, cmd->app); + forward_msg(manager, header, cmd->driver); manager_storage_subscribe(manager, cmd->app, cmd->driver, cmd->group, cmd->params); } else { @@ -1070,14 +1046,16 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) for (uint16_t i = 0; i < cmd->n_group; ++i) { neu_req_subscribe_group_info_t *info = &cmd->groups[i]; - error.error = neu_manager_subscribe(manager, cmd->app, info->driver, - info->group, info->params); + error.error = + neu_manager_subscribe(manager, cmd->app, info->driver, + info->group, info->params, &info->port); if (0 != error.error) { break; } - error.error = neu_manager_send_subscribe( - manager, cmd->app, info->driver, info->group, info->params); + error.error = neu_manager_send_subscribe(manager, cmd->app, + info->driver, info->group, + info->port, info->params); if (0 != error.error) { break; } @@ -1121,6 +1099,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error.error == NEU_ERR_SUCCESS) { forward_msg(manager, header, cmd->app); + forward_msg(manager, header, cmd->driver); manager_storage_unsubscribe(manager, cmd->app, cmd->driver, cmd->group); } @@ -1498,55 +1477,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) return 0; } -// inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, -// 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); -//} else { -// nlog_warn("forward msg to pipe %d fail", pipe.id); -// nng_msg_free(out_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); -// memcpy(ptr, cmd->tags[i].value.value.ptr.ptr, -// cmd->tags[i].value.value.ptr.length); -// 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, neu_reqresp_head_t *header, const char *node) { diff --git a/src/core/manager.h b/src/core/manager.h index 4984af51..8f7c8c25 100644 --- a/src/core/manager.h +++ b/src/core/manager.h @@ -29,4 +29,6 @@ neu_manager_t *neu_manager_create(); void neu_manager_destroy(neu_manager_t *manager); +uint16_t neu_manager_get_port(); + #endif diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 3a81494b..868384c6 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -683,7 +683,7 @@ static inline int manager_subscribe(neu_manager_t *manager, const char *app, int neu_manager_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, - const char *params) + const char *params, uint16_t *app_port) { if (neu_node_manager_is_monitor(manager->node_manager, app)) { // filter out monitor node @@ -696,6 +696,8 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app, return NEU_ERR_NODE_NOT_EXIST; } + *app_port = neu_adapter_trans_data_port(adapter); + // guard against empty mqtt topic parameter // this is not an elegant solution due to the current architecture if (params && 0 == strcmp(adapter->module->module_name, "MQTT")) { @@ -726,7 +728,7 @@ int neu_manager_update_subscribe(neu_manager_t *manager, const char *app, int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, - const char *params) + uint16_t app_port, const char *params) { memset(manager->buf, 0, sizeof(manager->buf)); neu_reqresp_head_t *header = (neu_reqresp_head_t *) manager->buf; @@ -738,6 +740,7 @@ int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, strcpy(cmd.app, app); strcpy(cmd.driver, driver); strcpy(cmd.group, group); + cmd.port = app_port; if (params && NULL == (cmd.params = strdup(params))) { return NEU_ERR_EINTERNAL; @@ -751,14 +754,29 @@ int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, int ret = sendto(manager->server_fd, header, header->len, 0, (struct sockaddr *) &addr, sizeof(addr)); if (ret != (int) header->len) { - nlog_warn("send %s to %s failed", neu_reqresp_type_string(header->type), - app); + nlog_warn("send %s to %s app failed", + neu_reqresp_type_string(header->type), app); free(cmd.params); } else { - nlog_notice("send %s to %s", neu_reqresp_type_string(header->type), + nlog_notice("send %s to %s app", neu_reqresp_type_string(header->type), app); } + strcpy(header->receiver, driver); + addr = neu_node_manager_get_addr(manager->node_manager, driver); + + neu_msg_gen(header, &cmd); + + ret = sendto(manager->server_fd, header, header->len, 0, + (struct sockaddr *) &addr, sizeof(addr)); + if (ret != (int) header->len) { + nlog_warn("send %s to %s driver failed", + neu_reqresp_type_string(header->type), app); + } else { + nlog_notice("send %s to %s driver", + neu_reqresp_type_string(header->type), app); + } + return 0; } diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 107c8221..0f5aef00 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -29,10 +29,9 @@ #include "template_manager.h" typedef struct neu_manager { - int server_fd; - uint8_t buf[NEU_MSG_MAX_SIZE]; - uint8_t recv_buf[NEU_MSG_MAX_SIZE]; - // nng_socket socket; + int server_fd; + uint8_t buf[NEU_MSG_MAX_SIZE]; + uint8_t recv_buf[NEU_MSG_MAX_SIZE]; neu_events_t * events; neu_event_io_t *loop; @@ -102,13 +101,13 @@ UT_array *neu_manager_get_driver_group(neu_manager_t *manager); int neu_manager_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, - const char *params); + const char *params, uint16_t *app_port); int neu_manager_update_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, const char *params); int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, - const char *params); + uint16_t app_port, const char *params); int neu_manager_unsubscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group); UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app); diff --git a/src/core/storage.c b/src/core/storage.c index 9caea1a0..bce44ea6 100644 --- a/src/core/storage.c +++ b/src/core/storage.c @@ -405,17 +405,18 @@ int manager_load_subscribe(neu_manager_t *manager) } else { utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info) { - rv = neu_manager_subscribe(manager, node->node, + uint16_t app_port = 0; + rv = neu_manager_subscribe(manager, node->node, info->driver_name, info->group_name, - info->params); + info->params, &app_port); const char *ok_or_err = (0 == rv) ? "success" : "fail"; nlog_notice("%s load subscription app:%s driver:%s grp:%s", ok_or_err, node->node, info->driver_name, info->group_name); if (0 == rv) { - neu_manager_send_subscribe(manager, node->node, - info->driver_name, - info->group_name, info->params); + neu_manager_send_subscribe( + manager, node->node, info->driver_name, + info->group_name, app_port, info->params); } } diff --git a/src/parser/neu_json_group_config.h b/src/parser/neu_json_group_config.h index 89536ff6..390bd062 100644 --- a/src/parser/neu_json_group_config.h +++ b/src/parser/neu_json_group_config.h @@ -135,9 +135,10 @@ int neu_json_decode_unsubscribe_req(char * buf, void neu_json_decode_unsubscribe_req_free(neu_json_unsubscribe_req_t *req); typedef struct { - char *driver; - char *group; - char *params; + char * driver; + char * group; + uint16_t port; + char * params; } neu_json_subscribe_groups_info_t; typedef struct {