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

feat(core): trans_data msg are sent using a separate channel

Browse Source
This commit is contained in:
fengzero
2023-10-08 06:59:07 +00:00
parent ff87cc73b6
commit bf4acdfb85
18 changed files with 666 additions and 643 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+160 -21
View File
@@ -24,8 +24,11 @@
extern "C" {
#endif
#include <pthread.h>
#include <stdint.h>
#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;
plugins/ekuiper/json_rw.c
+21 -125
View File
@@ -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));
plugins/ekuiper/plugin_ekuiper.c
-7
View File
@@ -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;
plugins/monitor/monitor.c
-2
View File
@@ -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));
plugins/mqtt/mqtt_handle.c
+18 -112
View File
@@ -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;
}
plugins/mqtt/mqtt_plugin.c
-10
View File
@@ -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;
plugins/restful/rest.c
-2
View File
@@ -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));
plugins/restful/rw_handle.c
+7 -105
View File
@@ -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);
}
src/adapter/adapter.c
+141 -23
View File
@@ -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;
src/adapter/adapter_internal.h
+11 -3
View File
@@ -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);
src/adapter/driver/driver.c
+245 -126
View File
@@ -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);
}
src/adapter/driver/driver_internal.h
+5
View File
@@ -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
src/core/manager.c
+18 -88
View File
@@ -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)
{
src/core/manager.h
+2
View File
@@ -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
src/core/manager_internal.c
+23 -5
View File
@@ -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;
}
src/core/manager_internal.h
+5 -6
View File
@@ -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);
src/core/storage.c
+6 -5
View File
@@ -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);
}
}
src/parser/neu_json_group_config.h
+4 -3
View File
@@ -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 {