Compare commits

...
52 Commits
Author SHA1 Message Date
fengzeroz 86a1fdf313 Merge pull request #2561 from fengzeroz/v2.12
version: 2.12.2
2025-09-05 13:57:59 +08:00
fengzero af8349f830 version: 2.12.2 2025-09-05 05:57:05 +00:00
fengzeroz 447b96e539 Merge pull request #2552 from fengzeroz/v2.12
conn: disconnect when connected and recv error
2025-08-27 14:43:07 +08:00
fengzero 3a4bd5eb9d conn: disconnect when connected and recv error 2025-08-27 06:40:28 +00:00
fengzeroz 45ab9227c1 Merge pull request #2548 from fengzeroz/v2.12
get subs: match sub string
2025-08-21 14:00:11 +08:00
fengzero c1835e11eb get subs: match sub string 2025-08-21 05:53:11 +00:00
fengzeroz e2ca35b62d Merge pull request #2545 from fengzeroz/v2.12
subs api: response ok when request empty
2025-08-21 11:04:13 +08:00
fengzero 1c1aa7a704 subs api: response ok when request empty 2025-08-21 03:00:01 +00:00
fengzeroz 23d4e7d43c Merge pull request #2542 from fengzeroz/v2.12
app: get all groups
2025-08-20 14:38:01 +08:00
fengzero d6d56500cd app: get all groups 2025-08-20 06:15:12 +00:00
fengzeroz 2f4b016124 Merge pull request #2536 from fengzeroz/v2.12
event name
2025-08-19 15:24:32 +08:00
fengzero ec46edbd86 event name 2025-08-19 07:01:14 +00:00
fengzeroz 3d641d8d26 Merge pull request #2532 from ZhouBox/v2.12
[v2.12]json:object array support
2025-08-08 15:53:27 +08:00
xiang.zhou 1efdc19b19 json:object array support 2025-08-08 15:02:10 +08:00
fengzeroz dc10430e7b Merge pull request #2524 from ZhouBox/v2.12
[v2.12]memery leak
2025-07-23 14:52:38 +08:00
xiang.zhou 6ac8443c59 memery leak 2025-07-23 14:31:07 +08:00
fengzeroz 001fbcc416 Merge pull request #2523 from ZhouBox/v2.12
[v2.12]memery leak
2025-07-23 13:47:58 +08:00
xiang.zhou 7d1c317fb6 memery leak 2025-07-23 12:04:23 +08:00
fengzeroz 5c49db7446 Merge pull request #2522 from ZhouBox/v2.12
[v2.12]fix: handle NULL json_str case to avoid undefined behavior
2025-07-22 15:21:04 +08:00
moruoxian f7abe05e72 fix: handle NULL json_str case to avoid undefined behavior 2025-07-22 14:43:00 +08:00
fengzeroz 9cca71b80a Merge pull request #2520 from fengzeroz/v2.12
version: 2.12.1
2025-07-17 09:47:33 +08:00
fengzero 6f9fd86457 version: 2.12.1 2025-07-17 01:44:28 +00:00
fengzeroz e65a6dd8cc Merge pull request #2517 from ZhouBox/v2.12
[v2.12]mem op when driver report callback
2025-07-08 14:18:39 +08:00
xiang.zhou 50b8da6f80 mem op when driver report callback 2025-07-08 13:54:21 +08:00
fengzeroz 15ec7713a4 Merge pull request #2512 from hxy7yx/v2.12-1
[v2.12]fix(datalayers):queue
2025-07-07 14:54:58 +08:00
hxy7yx 9371e834e8 fix(datalayers):queue 2025-06-26 19:04:20 -07:00
fengzeroz 816ddbe184 Merge pull request #2510 from fengzeroz/v2.12
report: pre alloc memory
2025-06-26 11:15:34 +08:00
fengzero 57b54d799c report: pre alloc memory 2025-06-26 03:13:45 +00:00
fengzeroz c83c31ddef Merge pull request #2498 from hxy7yx/v2.12-1
[v2.12]feat(modbus):support endianess of 8-Byte data
2025-06-16 11:23:39 +08:00
fengzeroz 27a68ad136 Merge pull request #2494 from fengzeroz/v2.12
fix: mqtt disable subscribe topic
2025-06-16 11:23:28 +08:00
fengzero f58eb9f81e fix cppcheck 2025-06-16 03:14:02 +00:00
hxy7yx d4c82bf179 feat(modbus):support endianess of 8-Byte data 2025-06-15 19:46:27 -07:00
fengzero fe32656109 fix: mqtt disable subscribe topic 2025-06-16 02:31:59 +00:00
fengzeroz f8927e6ad4 Merge pull request #2487 from fengzeroz/v2.12
mqtt: rm / in topic
2025-06-12 11:23:15 +08:00
fengzero 3823226b67 mqtt: rm / in topic 2025-06-12 03:09:31 +00:00
fengzeroz 05d043f729 Merge pull request #2484 from fengzeroz/v2.12
mqtt: config enable_topic
2025-06-12 11:08:42 +08:00
fengzero 8ee7ab8a36 mqtt: config enable_topic 2025-06-12 02:57:08 +00:00
fengzeroz f0b746a1ab Merge pull request #2482 from fengzeroz/v2.12
fix: filter error crash
2025-06-11 15:38:32 +08:00
fengzero afaddb8066 fix: filter error crash 2025-06-11 03:02:38 +00:00
fengzeroz 9957771212 Merge pull request #2478 from ZhouBox/v2.12
[v2.12]startup optimization
2025-06-11 10:19:56 +08:00
xiang.zhou 89e1a7da3f startup optimization 2025-06-10 14:37:04 +08:00
fengzeroz 41c737ba4b Merge pull request #2474 from ZhouBox/v2.12
[v2.12]adapter:check state when do with trans data
2025-06-09 16:35:37 +08:00
xiang.zhou 6a7bea735f adapter:check state when do with trans data 2025-06-09 16:24:50 +08:00
fengzeroz 20098f3845 Merge pull request #2469 from hxy7yx/v2.12-1
start datalayers plugin without check
2025-05-29 09:41:04 +08:00
hxy7yx 91c3b39e5f start datalayers plugin without check 2025-05-27 20:36:39 -07:00
fengzeroz 50771e55b4 Merge pull request #2466 from hxy7yx/v2.12-1
datalayers:cancel verifying connections before configuration
2025-05-28 10:04:21 +08:00
hxy7yx d1abf23335 datalayers:cancel verifying connections before configuration 2025-05-27 00:12:25 -07:00
fengzeroz 552827dc16 Merge pull request #2463 from hxy7yx/v2.12-1
[v2.12]update ci.yml
2025-05-26 17:58:28 +08:00
hxy7yx 86271ecca8 update api:sub info of datalayers plugin 2025-05-26 02:49:30 -07:00
hxy7yx 9db3a8e521 update ci.yml 2025-05-25 20:10:51 -07:00
fengzeroz 7e2ee821b2 Merge pull request #2460 from fengzeroz/v2.12
branch v2.12
2025-05-23 14:51:37 +08:00
fengzero 39564dce45 branch v2.12 2025-05-23 06:51:13 +00:00
61 changed files with 1313 additions and 427 deletions
+4 -4
View File
@@ -27,7 +27,7 @@ jobs:
cppcheck:
runs-on: ubuntu-24.04
container: ghcr.io/neugates/build:x86_64-main
container: ghcr.io/neugates/build:x86_64-v2.12
steps:
- uses: actions/checkout@v4
@@ -42,7 +42,7 @@ jobs:
build:
runs-on: ubuntu-24.04
container: ghcr.io/neugates/build:x86_64-main
container: ghcr.io/neugates/build:x86_64-v2.12
steps:
- uses: actions/checkout@v4
@@ -65,7 +65,7 @@ jobs:
ut:
runs-on: ubuntu-24.04
container: ghcr.io/neugates/build:x86_64-main
container: ghcr.io/neugates/build:x86_64-v2.12
needs: [build]
steps:
@@ -103,7 +103,7 @@ jobs:
pft:
runs-on: ubuntu-24.04
container: ghcr.io/neugates/build:x86_64-main
container: ghcr.io/neugates/build:x86_64-v2.12
needs: [build]
strategy:
matrix:
+4 -4
View File
@@ -8,9 +8,9 @@ on:
- published
env:
VERSION: 2.12.0
VERSION: 2.12.2
UI_VERSION: 2.6.1
BRANCH: main
BRANCH: v2.12
jobs:
cross-build:
@@ -19,7 +19,7 @@ jobs:
matrix:
arch:
- [x86_64-linux-gnu, x86_64, amd64]
container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.12
steps:
- uses: actions/checkout@v4
@@ -62,7 +62,7 @@ jobs:
matrix:
arch:
- [x86_64-linux-gnu, x86_64, amd64]
container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.12
steps:
- uses: actions/checkout@v4
+1 -1
View File
@@ -1,7 +1,7 @@
set(CMAKE_SYSTEM_NAME Linux)
set(COMPILER_PREFIX aarch64-linux-gnu)
set(CMAKE_SYSTEM_PROCESSOR aarch64)
set(LIBRARY_DIR /home/neuron/main/libs)
set(LIBRARY_DIR /home/neuron/v2.12/libs)
set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+1 -1
View File
@@ -1,7 +1,7 @@
set(CMAKE_SYSTEM_NAME Linux)
set(COMPILER_PREFIX arm-linux-gnueabi)
set(CMAKE_SYSTEM_PROCESSOR armv5tejl)
set(LIBRARY_DIR /home/neuron/main/libs)
set(LIBRARY_DIR /home/neuron/v2.12/libs)
set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+1 -1
View File
@@ -1,7 +1,7 @@
set(CMAKE_SYSTEM_NAME Linux)
set(COMPILER_PREFIX arm-linux-gnueabihf)
set(CMAKE_SYSTEM_PROCESSOR armv7l)
set(LIBRARY_DIR /home/neuron/main/libs)
set(LIBRARY_DIR /home/neuron/v2.12/libs)
set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+1 -1
View File
@@ -1,7 +1,7 @@
set(CMAKE_SYSTEM_NAME Linux)
set(COMPILER_PREFIX x86_64-linux-gnu)
set(CMAKE_SYSTEM_PROCESSOR x86_64)
set(LIBRARY_DIR /home/neuron/main/libs)
set(LIBRARY_DIR /home/neuron/v2.12/libs)
set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+1 -1
View File
@@ -24,7 +24,7 @@
#define NEU_VERSION_MAJOR 2
#define NEU_VERSION_MINOR 12
#define NEU_VERSION_FIX 0
#define NEU_VERSION_FIX 2
#define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
#define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)
+1 -1
View File
@@ -35,7 +35,7 @@ typedef struct neu_events neu_events_t;
* thread.
* @return the newly created event.
*/
neu_events_t *neu_event_new(void);
neu_events_t *neu_event_new(const char *name);
/**
* @brief Close a event.
+87 -33
View File
@@ -61,6 +61,8 @@ typedef enum neu_reqresp_type {
NEU_REQ_SUBSCRIBE_GROUPS,
NEU_REQ_GET_SUBSCRIBE_GROUP,
NEU_RESP_GET_SUBSCRIBE_GROUP,
NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP,
NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP,
NEU_REQ_GET_SUB_DRIVER_TAGS,
NEU_RESP_GET_SUB_DRIVER_TAGS,
@@ -68,6 +70,8 @@ typedef enum neu_reqresp_type {
NEU_RESP_GET_DATALAYERS_GROUPS,
NEU_REQ_GET_DATALAYERS_TAGS,
NEU_RESP_GET_DATALAYERS_TAGS,
NEU_REQ_GET_DATALAYERS_TAG,
NEU_RESP_GET_DATALAYERS_TAG,
NEU_REQ_NODE_INIT,
NEU_REQ_NODE_UNINIT,
@@ -186,19 +190,24 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_RESP_WRITE_TAGS] = "NEU_RESP_WRITE_TAGS",
[NEU_REQ_WRITE_GTAGS] = "NEU_REQ_WRITE_GTAGS",
[NEU_REQ_SUBSCRIBE_GROUP] = "NEU_REQ_SUBSCRIBE_GROUP",
[NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP",
[NEU_REQ_UPDATE_SUBSCRIBE_GROUP] = "NEU_REQ_UPDATE_SUBSCRIBE_GROUP",
[NEU_REQ_SUBSCRIBE_GROUPS] = "NEU_REQ_SUBSCRIBE_GROUPS",
[NEU_REQ_GET_SUBSCRIBE_GROUP] = "NEU_REQ_GET_SUBSCRIBE_GROUP",
[NEU_RESP_GET_SUBSCRIBE_GROUP] = "NEU_RESP_GET_SUBSCRIBE_GROUP",
[NEU_REQ_GET_SUB_DRIVER_TAGS] = "NEU_REQ_GET_SUB_DRIVER_TAGS",
[NEU_RESP_GET_SUB_DRIVER_TAGS] = "NEU_RESP_GET_SUB_DRIVER_TAGS",
[NEU_REQ_SUBSCRIBE_GROUP] = "NEU_REQ_SUBSCRIBE_GROUP",
[NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP",
[NEU_REQ_UPDATE_SUBSCRIBE_GROUP] = "NEU_REQ_UPDATE_SUBSCRIBE_GROUP",
[NEU_REQ_SUBSCRIBE_GROUPS] = "NEU_REQ_SUBSCRIBE_GROUPS",
[NEU_REQ_GET_SUBSCRIBE_GROUP] = "NEU_REQ_GET_SUBSCRIBE_GROUP",
[NEU_RESP_GET_SUBSCRIBE_GROUP] = "NEU_RESP_GET_SUBSCRIBE_GROUP",
[NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP] = "NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP",
[NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP] =
"NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP",
[NEU_REQ_GET_SUB_DRIVER_TAGS] = "NEU_REQ_GET_SUB_DRIVER_TAGS",
[NEU_RESP_GET_SUB_DRIVER_TAGS] = "NEU_RESP_GET_SUB_DRIVER_TAGS",
[NEU_REQ_GET_DATALAYERS_GROUPS] = "NEU_REQ_GET_DATALAYERS_GROUPS",
[NEU_RESP_GET_DATALAYERS_GROUPS] = "NEU_RESP_GET_DATALAYERS_GROUPS",
[NEU_REQ_GET_DATALAYERS_TAGS] = "NEU_REQ_GET_DATALAYERS_TAGS",
[NEU_RESP_GET_DATALAYERS_TAGS] = "NEU_RESP_GET_DATALAYERS_TAGS",
[NEU_REQ_GET_DATALAYERS_TAG] = "NEU_REQ_GET_DATALAYERS_TAG",
[NEU_RESP_GET_DATALAYERS_TAG] = "NEU_RESP_GET_DATALAYERS_TAG",
[NEU_REQ_NODE_INIT] = "NEU_REQ_NODE_INIT",
[NEU_REQ_NODE_UNINIT] = "NEU_REQ_NODE_UNINIT",
@@ -695,8 +704,16 @@ typedef struct neu_req_get_subscribe_group {
char group[NEU_GROUP_NAME_LEN];
} neu_req_get_subscribe_group_t;
typedef struct neu_req_get_driver_subscribe_group {
char app[NEU_NODE_NAME_LEN];
char name[NEU_NODE_NAME_LEN];
} neu_req_get_driver_subscribe_group_t;
typedef struct {
char app[NEU_NODE_NAME_LEN];
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char tag[NEU_TAG_NAME_LEN];
} neu_req_get_sub_driver_tags_t;
typedef struct neu_resp_subscribe_info {
@@ -707,6 +724,13 @@ typedef struct neu_resp_subscribe_info {
char *static_tags;
} neu_resp_subscribe_info_t;
typedef struct neu_resp_driver_subscribe_info {
char driver[NEU_NODE_NAME_LEN];
char app[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
bool subscribed;
} neu_resp_driver_subscribe_info_t;
static inline void neu_resp_subscribe_info_fini(neu_resp_subscribe_info_t *info)
{
free(info->params);
@@ -730,7 +754,8 @@ typedef struct {
} neu_resp_get_sub_driver_tags_t;
typedef struct neu_resp_get_subscribe_group {
UT_array *groups; // array neu_resp_subscribe_info_t
UT_array *groups; // array neu_resp_subscribe_info_t or
// neu_resp_driver_subscribe_info_t
} neu_resp_get_subscribe_group_t;
typedef struct neu_req_node_setting {
@@ -881,10 +906,11 @@ typedef struct neu_resp_tag_value {
typedef neu_resp_tag_value_t neu_tag_value_t;
typedef struct neu_resp_tag_value_meta {
char tag[NEU_TAG_NAME_LEN];
neu_dvalue_t value;
neu_tag_meta_t metas[NEU_TAG_META_SIZE];
neu_datatag_t datatag;
char tag[NEU_TAG_NAME_LEN];
neu_dvalue_t value;
neu_tag_meta_t *metas;
int n_meta;
neu_datatag_t datatag;
} neu_resp_tag_value_meta_t;
static inline UT_icd *neu_resp_tag_value_meta_icd()
@@ -955,10 +981,11 @@ typedef struct {
} neu_resp_read_group_t;
typedef struct neu_resp_tag_value_meta_paginate {
char tag[NEU_TAG_NAME_LEN];
neu_dvalue_t value;
neu_tag_meta_t metas[NEU_TAG_META_SIZE];
neu_datatag_t datatag;
char tag[NEU_TAG_NAME_LEN];
neu_dvalue_t value;
neu_tag_meta_t *metas;
int n_meta;
neu_datatag_t datatag;
} neu_resp_tag_value_meta_paginate_t;
static inline UT_icd *neu_resp_tag_value_meta_paginate_icd()
@@ -989,6 +1016,13 @@ static inline void neu_resp_read_free(neu_resp_read_group_t *resp)
tag_value->value.type < NEU_TYPE_ARRAY_STRING) {
free(tag_value->value.value.bools.bools);
}
if (tag_value->metas != NULL) {
for (int i = 0; i < tag_value->n_meta; i++) {
neu_free_dvalue(&tag_value->metas[i].value);
}
free(tag_value->metas);
}
}
free(resp->driver);
free(resp->group);
@@ -1006,6 +1040,13 @@ neu_resp_read_paginate_free(neu_resp_read_group_paginate_t *resp)
tag_value->value.type < NEU_TYPE_ARRAY_STRING) {
free(tag_value->value.value.bools.bools);
}
if (tag_value->metas != NULL) {
for (int i = 0; i < tag_value->n_meta; i++) {
neu_free_dvalue(&tag_value->metas[i].value);
}
free(tag_value->metas);
}
}
free(resp->driver);
free(resp->group);
@@ -1083,6 +1124,13 @@ static inline void neu_trans_data_free(neu_reqresp_trans_data_t *data)
tag_value->value.type < NEU_TYPE_ARRAY_STRING) {
free(tag_value->value.value.bools.bools);
}
if (tag_value->metas != NULL) {
for (int i = 0; i < tag_value->n_meta; i++) {
neu_free_dvalue(&tag_value->metas[i].value);
}
free(tag_value->metas);
}
}
utarray_free(data->tags);
free(data->group);
@@ -1101,18 +1149,21 @@ static inline void neu_tag_value_to_json(neu_resp_tag_value_meta_t *tag_value,
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;
}
}
// for (int k = 0; k < NEU_TAG_META_SIZE; k++) {
// if (strlen(tag_value->metas[k].name) > 0) {
// tag_json->n_meta++;
// } else {
// break;
// }
// }
tag_json->n_meta = tag_value->n_meta;
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);
neu_json_metas_to_json(tag_value->metas, tag_value->n_meta, tag_json);
tag_json->datatag.bias = tag_value->datatag.bias;
@@ -1310,18 +1361,21 @@ neu_tag_value_to_json_paginate(neu_resp_tag_value_meta_paginate_t *tag_value,
memcpy(tag_json->datatag.meta, tag_value->datatag.meta,
NEU_TAG_META_LENGTH);
for (int k = 0; k < NEU_TAG_META_SIZE; k++) {
if (strlen(tag_value->metas[k].name) > 0) {
tag_json->n_meta++;
} else {
break;
}
}
// for (int k = 0; k < NEU_TAG_META_SIZE; k++) {
// if (strlen(tag_value->metas[k].name) > 0) {
// tag_json->n_meta++;
// } else {
// break;
// }
// }
tag_json->n_meta = tag_value->n_meta;
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_paginate(tag_value->metas, NEU_TAG_META_SIZE,
neu_json_metas_to_json_paginate(tag_value->metas, tag_value->n_meta,
tag_json);
switch (tag_value->value.type) {
+2
View File
@@ -42,6 +42,8 @@ neu_otel_trace_ctx neu_otel_find_trace2(void *req_ctx);
void neu_otel_trace_set_final(neu_otel_trace_ctx ctx);
void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num);
void neu_otel_trace_reduce_expected_span_num(neu_otel_trace_ctx ctx,
uint32_t num);
uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx);
neu_otel_trace_ctx neu_otel_find_trace_by_id(const char *trace_id);
+5
View File
@@ -43,6 +43,10 @@ typedef enum {
NEU_DATATAG_ENDIAN_BL32 = 5, // #BL 2,1,4,3
NEU_DATATAG_ENDIAN_L64 = 6, // #L 8,7,6,5,4,3,2,1
NEU_DATATAG_ENDIAN_B64 = 7, // #B 1,2,3,4,5,6,7,8
NEU_DATATAG_ENDIAN_LL64 = 8,
NEU_DATATAG_ENDIAN_LB64 = 9,
NEU_DATATAG_ENDIAN_BB64 = 10,
NEU_DATATAG_ENDIAN_BL64 = 11,
} neu_datatag_endian_e;
typedef enum {
@@ -62,6 +66,7 @@ typedef union {
} value32;
struct {
neu_datatag_endian_e endian;
bool is_default;
} value64;
struct {
uint16_t length;
+36
View File
@@ -890,6 +890,42 @@ static inline uint64_t neu_htonll(uint64_t u)
return ret;
}
static inline uint64_t neu_htonlb(uint64_t u)
{
uint8_t *in = (uint8_t *) &u;
uint64_t out = 0;
uint8_t *out_bytes = (uint8_t *) &out;
out_bytes[0] = in[1];
out_bytes[1] = in[0];
out_bytes[2] = in[3];
out_bytes[3] = in[2];
out_bytes[4] = in[5];
out_bytes[5] = in[4];
out_bytes[6] = in[7];
out_bytes[7] = in[6];
return out;
}
static inline uint64_t neu_htonbl(uint64_t u)
{
uint8_t *in = (uint8_t *) &u;
uint64_t out = 0;
uint8_t *out_bytes = (uint8_t *) &out;
out_bytes[0] = in[6];
out_bytes[1] = in[7];
out_bytes[2] = in[4];
out_bytes[3] = in[5];
out_bytes[4] = in[2];
out_bytes[5] = in[3];
out_bytes[6] = in[0];
out_bytes[7] = in[1];
return out;
}
static inline uint64_t neu_ntohll(uint64_t u)
{
return neu_htonll(u);
+17 -8
View File
@@ -185,21 +185,27 @@ void db_write_task_consumer(neu_plugin_t *plugin)
db_write_task_t *task = NULL;
while (1) {
pthread_rwlock_rdlock(&plugin->plugin_mutex);
if (plugin->consumer_thread_stop_flag) {
pthread_rwlock_unlock(&plugin->plugin_mutex);
pthread_mutex_lock(&plugin->queue_mutex);
while (plugin->task_queue.size == 0 &&
!plugin->consumer_thread_stop_flag) {
pthread_cond_wait(&plugin->queue_not_empty, &plugin->queue_mutex);
}
if (plugin->consumer_thread_stop_flag && plugin->task_queue.size == 0) {
pthread_mutex_unlock(&plugin->queue_mutex);
break;
}
pthread_rwlock_unlock(&plugin->plugin_mutex);
task = task_queue_pop(plugin, &plugin->task_queue);
pthread_mutex_unlock(&plugin->queue_mutex);
if (task) {
db_write_task_cb(task, plugin);
task_free(task);
} else {
usleep(1000);
}
}
plog_notice(plugin, "Consumer thread exiting gracefully.\n");
}
@@ -558,10 +564,13 @@ int handle_trans_data(neu_plugin_t * plugin,
task->bool_tags = bool_tags;
task->string_tags = string_tags;
task_queue_push(plugin, task);
pthread_rwlock_unlock(&plugin->plugin_mutex);
pthread_mutex_lock(&plugin->queue_mutex);
task_queue_push(plugin, task);
pthread_cond_signal(&plugin->queue_not_empty);
pthread_mutex_unlock(&plugin->queue_mutex);
return rv;
}
+3
View File
@@ -68,6 +68,9 @@ struct neu_plugin {
task_queue_t task_queue;
pthread_t consumer_thread;
pthread_mutex_t queue_mutex;
pthread_cond_t queue_not_empty;
pthread_rwlock_t plugin_mutex;
bool consumer_thread_stop_flag;
+22 -8
View File
@@ -33,9 +33,9 @@ extern const neu_plugin_module_t neu_plugin_module;
void create_consumer_thread(neu_plugin_t *plugin)
{
pthread_rwlock_wrlock(&plugin->plugin_mutex);
pthread_mutex_lock(&plugin->queue_mutex);
plugin->consumer_thread_stop_flag = false;
pthread_rwlock_unlock(&plugin->plugin_mutex);
pthread_mutex_unlock(&plugin->queue_mutex);
int status = pthread_create(&plugin->consumer_thread, NULL,
(void *) db_write_task_consumer, plugin);
@@ -54,9 +54,10 @@ neu_plugin_t *datalayers_plugin_open(void)
void stop_consumer_thread(pthread_t *consumer_thread, neu_plugin_t *plugin)
{
pthread_rwlock_wrlock(&plugin->plugin_mutex);
pthread_mutex_lock(&plugin->queue_mutex);
plugin->consumer_thread_stop_flag = true;
pthread_rwlock_unlock(&plugin->plugin_mutex);
pthread_cond_broadcast(&plugin->queue_not_empty);
pthread_mutex_unlock(&plugin->queue_mutex);
int join_status = pthread_join(*consumer_thread, NULL);
if (join_status != 0) {
@@ -81,11 +82,17 @@ int datalayers_plugin_init(neu_plugin_t *plugin, bool load)
pthread_rwlock_init(&plugin->plugin_mutex, NULL);
pthread_mutex_init(&plugin->queue_mutex, NULL);
pthread_cond_init(&plugin->queue_not_empty, NULL);
pthread_mutex_lock(&plugin->queue_mutex);
plugin->consumer_thread_stop_flag = false;
pthread_mutex_unlock(&plugin->queue_mutex);
task_queue_init(&plugin->task_queue);
pthread_create(&plugin->consumer_thread, NULL,
(void *) db_write_task_consumer, plugin);
pthread_detach(plugin->consumer_thread);
NEU_PLUGIN_REGISTER_CACHED_QUEUE_SIZE_METRIC(plugin);
NEU_PLUGIN_REGISTER_MAX_CACHED_QUEUE_SIZE_METRIC(plugin);
@@ -101,7 +108,11 @@ int datalayers_plugin_uninit(neu_plugin_t *plugin)
pthread_rwlock_wrlock(&plugin->plugin_mutex);
stop_consumer_thread(&plugin->consumer_thread, plugin);
pthread_mutex_lock(&plugin->queue_mutex);
tasks_free(&plugin->task_queue);
pthread_mutex_unlock(&plugin->queue_mutex);
pthread_mutex_destroy(&plugin->queue_mutex);
pthread_cond_destroy(&plugin->queue_not_empty);
datalayers_config_fini(&plugin->config);
route_tbl_free(plugin->route_tbl);
@@ -135,10 +146,10 @@ static int config_datalayers_client(neu_plugin_t * plugin,
if (NULL == plugin->client) {
plog_error(plugin, "datalayers client_create failed");
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
return NEU_ERR_DATALAYERS_CONNECT_FAILURE;
} else {
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
}
plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
return NEU_ERR_SUCCESS;
}
@@ -149,7 +160,9 @@ int datalayers_plugin_config(neu_plugin_t *plugin, const char *setting)
datalayers_config_t config = { 0 };
if (plugin->client != NULL) {
stop_consumer_thread(&plugin->consumer_thread, plugin);
pthread_mutex_lock(&plugin->queue_mutex);
tasks_free(&plugin->task_queue);
pthread_mutex_unlock(&plugin->queue_mutex);
client_destroy(plugin->client);
plugin->client = NULL;
create_consumer_thread(plugin);
@@ -185,7 +198,8 @@ int datalayers_plugin_start(neu_plugin_t *plugin)
int ret = config_datalayers_client(plugin, &plugin->config);
if (ret != 0) {
plog_error(plugin, "datalayers started failed");
return NEU_ERR_DATALAYERS_CONNECT_FAILURE;
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
return NEU_ERR_SUCCESS;
}
}
+34 -5
View File
@@ -18,11 +18,11 @@
},
{
"type": 7,
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
},
{
"type": 8,
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
},
{
"type": 9,
@@ -30,7 +30,7 @@
},
{
"type": 10,
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
},
{
"type": 11,
@@ -158,8 +158,8 @@
}
},
"endianess": {
"name": "Endianess",
"name_zh": "字节序",
"name": "Endianess of 4-Byte Data",
"name_zh": "4 字节数据字节序",
"description": "Tag byte order, ABCD corresponds to 1234",
"description_zh": "点位字节序,ABCD 对应 1234",
"attribute": "required",
@@ -186,6 +186,35 @@
]
}
},
"endianess_64": {
"name": "Endianess of 8-Byte Data",
"name_zh": "8 字节数据字节序",
"description": "Tag byte order, applicable to int64, uint64 and double types",
"description_zh": "点位字节序,适用于 int64,uint64 和 double 类型",
"attribute": "required",
"type": "map",
"default": 1,
"valid": {
"map": [
{
"key": "12 34 56 78",
"value": 1
},
{
"key": "21 43 65 87",
"value": 2
},
{
"key": "87 65 43 21",
"value": 3
},
{
"key": "78 56 34 12",
"value": 4
}
]
}
},
"address_base": {
"name": "Start Address",
"name_zh": "开始地址",
+34 -5
View File
@@ -18,11 +18,11 @@
},
{
"type": 7,
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
},
{
"type": 8,
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
},
{
"type": 9,
@@ -30,7 +30,7 @@
},
{
"type": 10,
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
"regex": "^[0-9]+![3-4][0-9]+(#B|#L|#BB|#BL|#LL|#LB|)$"
},
{
"type": 11,
@@ -166,8 +166,8 @@
}
},
"endianess": {
"name": "Endianess",
"name_zh": "字节序",
"name": "Endianess of 4-Byte Data",
"name_zh": "4 字节数据字节序",
"description": "Tag byte order, ABCD corresponds to 1234",
"description_zh": "点位字节序,ABCD 对应 1234",
"attribute": "required",
@@ -194,6 +194,35 @@
]
}
},
"endianess_64": {
"name": "Endianess of 8-Byte Data",
"name_zh": "8 字节数据字节序",
"description": "Tag byte order, applicable to int64, uint64 and double types",
"description_zh": "点位字节序,适用于 int64,uint64 和 double 类型",
"attribute": "required",
"type": "map",
"default": 1,
"valid": {
"map": [
{
"key": "12 34 56 78",
"value": 1
},
{
"key": "21 43 65 87",
"value": 2
},
{
"key": "87 65 43 21",
"value": 3
},
{
"key": "78 56 34 12",
"value": 4
}
]
}
},
"address_base": {
"name": "Start Address",
"name_zh": "开始地址",
+7
View File
@@ -73,6 +73,13 @@ typedef enum modbus_endianess {
MODBUS_CDAB = 4,
} modbus_endianess;
typedef enum modbus_endianess_64 {
MODBUS_LL = 1, // 12 34 56 78
MODBUS_LB = 2, // 21 43 65 87
MODBUS_BB = 3, // 87 65 43 21
MODBUS_BL = 4, // 78 56 34 12
} modbus_endianess_64;
typedef enum modbus_address_base {
base_0 = 0,
base_1 = 1,
+36 -10
View File
@@ -238,7 +238,8 @@ modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte)
}
int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option,
modbus_endianess endianess, bool default_tag_endian)
modbus_endianess endianess, bool default_tag_endian,
modbus_endianess_64 endianess_64, bool default_tag_endian_64)
{
int n = 0;
switch (type) {
@@ -260,6 +261,9 @@ int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option,
case NEU_TYPE_DOUBLE:
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
if (default_tag_endian_64) {
modbus_convert_endianess_64(value, endianess_64);
}
n = sizeof(uint64_t);
value->u64 = neu_htonll(value->u64);
break;
@@ -293,8 +297,9 @@ int cal_n_byte(int type, neu_value_u *value, neu_datatag_addr_option_u option,
return n;
}
modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array * tags,
modbus_endianess endianess)
modbus_write_cmd_sort_t *
modbus_write_tags_sort(UT_array *tags, modbus_endianess endianess,
modbus_endianess_64 endianess_64)
{
neu_tag_sort_result_t *result =
neu_tag_sort(tags, tag_sort_write, tag_cmp_write);
@@ -316,17 +321,18 @@ modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array * tags,
utarray_foreach(result->sorts[i].tags, modbus_point_write_t **, tag_s)
{
if ((*tag_s)->point.area == MODBUS_AREA_COIL) {
n_byte_tag =
cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
(*tag_s)->point.option, endianess, true);
n_byte_tag = cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
(*tag_s)->point.option, endianess, true,
endianess_64, true);
data_bit[k / 8] += ((*tag_s)->value.i8) << k % 8;
n_byte += n_byte_tag;
k++;
} else {
n_byte_tag =
cal_n_byte((*tag_s)->point.type, &(*tag_s)->value,
(*tag_s)->point.option, endianess,
(*tag_s)->point.option.value32.is_default);
n_byte_tag = cal_n_byte(
(*tag_s)->point.type, &(*tag_s)->value,
(*tag_s)->point.option, endianess,
(*tag_s)->point.option.value32.is_default, endianess_64,
(*tag_s)->point.option.value64.is_default);
memcpy(sort_result->cmd[i].bytes +
2 *
((*tag_s)->point.start_address -
@@ -558,4 +564,24 @@ void modbus_convert_endianess(neu_value_u *value, modbus_endianess endianess)
default:
break;
}
}
void modbus_convert_endianess_64(neu_value_u * value,
modbus_endianess_64 endianess_64)
{
switch (endianess_64) {
case MODBUS_LL:
break;
case MODBUS_BB:
value->u64 = neu_htonll(value->u64);
break;
case MODBUS_LB:
value->u64 = neu_htonlb(value->u64);
break;
case MODBUS_BL:
value->u64 = neu_htonbl(value->u64);
break;
default:
break;
}
}
+7 -4
View File
@@ -81,12 +81,15 @@ typedef struct modbus_write_cmd_sort {
modbus_write_cmd_t *cmd;
} modbus_write_cmd_sort_t;
modbus_read_cmd_sort_t * modbus_tag_sort(UT_array *tags, uint16_t max_byte);
modbus_write_cmd_sort_t *modbus_write_tags_sort(UT_array * tags,
modbus_endianess endianess);
void modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
modbus_read_cmd_sort_t *modbus_tag_sort(UT_array *tags, uint16_t max_byte);
modbus_write_cmd_sort_t *
modbus_write_tags_sort(UT_array *tags, modbus_endianess endianess,
modbus_endianess_64 endianess_64);
void modbus_tag_sort_free(modbus_read_cmd_sort_t *cs);
void modbus_convert_endianess(neu_value_u *value, modbus_endianess endianess);
void modbus_convert_endianess_64(neu_value_u * value,
modbus_endianess_64 endianess_64);
#ifdef __cplusplus
}
+9 -1
View File
@@ -472,6 +472,10 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
case NEU_TYPE_DOUBLE:
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
if ((*p_tag)->option.value64.is_default) {
modbus_convert_endianess_64(&dvalue.value,
plugin->endianess_64);
}
dvalue.value.u64 = neu_ntohll(dvalue.value.u64);
break;
case NEU_TYPE_BIT: {
@@ -729,6 +733,9 @@ static uint8_t convert_value(neu_plugin_t *plugin, neu_value_u *value,
case NEU_TYPE_DOUBLE:
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
if (tag->option.value64.is_default) {
modbus_convert_endianess_64(value, plugin->endianess_64);
}
value->u64 = neu_htonll(value->u64);
n_byte = sizeof(uint64_t);
break;
@@ -821,7 +828,8 @@ int modbus_write_tags(neu_plugin_t *plugin, void *req, UT_array *tags)
utarray_push_back(gtags->tags, &p);
}
gtags->cmd_sort = modbus_write_tags_sort(gtags->tags, plugin->endianess);
gtags->cmd_sort = modbus_write_tags_sort(gtags->tags, plugin->endianess,
plugin->endianess_64);
for (uint16_t i = 0; i < gtags->cmd_sort->n_cmd; i++) {
ret = write_modbus_points(plugin, &gtags->cmd_sort->cmd[i], req);
if (ret <= 0) {
+1
View File
@@ -40,6 +40,7 @@ struct neu_plugin {
modbus_protocol_e protocol;
modbus_endianess endianess;
modbus_endianess_64 endianess_64;
modbus_address_base address_base;
uint16_t interval;
+11 -1
View File
@@ -107,7 +107,7 @@ static int driver_init(neu_plugin_t *plugin, bool load)
{
(void) load;
plugin->protocol = MODBUS_PROTOCOL_RTU;
plugin->events = neu_event_new();
plugin->events = neu_event_new(plugin->common.name);
plugin->stack = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_RTU,
modbus_send_msg, modbus_value_handle,
modbus_write_resp);
@@ -183,6 +183,9 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
neu_json_elem_t address_base = { .name = "address_base",
.t = NEU_JSON_INT };
neu_json_elem_t endianess_64 = { .name = "endianess_64",
.t = NEU_JSON_INT };
ret = neu_parse_param((char *) config, &err_param, 3, &link, &timeout,
&interval);
if (ret != 0) {
@@ -244,6 +247,12 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
address_base.v.val_int = base_1;
}
ret = neu_parse_param((char *) config, &err_param, 1, &endianess_64);
if (ret != 0) {
free(err_param);
endianess_64.v.val_int = MODBUS_LL;
}
param.log = plugin->common.log;
plugin->max_retries = max_retries.v.val_int;
plugin->retry_interval = retry_interval.v.val_int;
@@ -252,6 +261,7 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
plugin->degrade_time = degrade_time.v.val_int;
plugin->endianess = endianess.v.val_int;
plugin->address_base = address_base.v.val_int;
plugin->endianess_64 = endianess_64.v.val_int;
if (link.v.val_int == 0) {
param.type = NEU_CONN_TTY_CLIENT;
+10 -1
View File
@@ -110,7 +110,7 @@ static int driver_init(neu_plugin_t *plugin, bool load)
{
(void) load;
plugin->protocol = MODBUS_PROTOCOL_TCP;
plugin->events = neu_event_new();
plugin->events = neu_event_new(plugin->common.name);
plugin->stack = modbus_stack_create((void *) plugin, MODBUS_PROTOCOL_TCP,
modbus_send_msg, modbus_value_handle,
modbus_write_resp);
@@ -190,6 +190,8 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
neu_json_elem_t endianess = { .name = "endianess", .t = NEU_JSON_INT };
neu_json_elem_t address_base = { .name = "address_base",
.t = NEU_JSON_INT };
neu_json_elem_t endianess_64 = { .name = "endianess_64",
.t = NEU_JSON_INT };
neu_json_elem_t backup_ip = { .name = "backup_host",
.t = NEU_JSON_STR,
@@ -260,6 +262,12 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
backup = true;
}
ret = neu_parse_param((char *) config, &err_param, 1, &endianess_64);
if (ret != 0) {
free(err_param);
endianess_64.v.val_int = MODBUS_LL;
}
param.log = plugin->common.log;
param_backup.log = plugin->common.log;
plugin->interval = interval.v.val_int;
@@ -270,6 +278,7 @@ static int driver_config(neu_plugin_t *plugin, const char *config)
plugin->degrade_cycle = degrade_cycle.v.val_int;
plugin->degrade_time = degrade_time.v.val_int;
plugin->endianess = endianess.v.val_int;
plugin->endianess_64 = endianess_64.v.val_int;
plugin->address_base = address_base.v.val_int;
if (mode.v.val_int == 1) {
+1 -1
View File
@@ -137,7 +137,7 @@ static int start_hearbeat_timer(neu_plugin_t *plugin, uint64_t interval)
}
if (NULL == plugin->events) {
plugin->events = neu_event_new();
plugin->events = neu_event_new(plugin->common.name);
if (NULL == plugin->events) {
plog_error(plugin, "neu_event_new fail");
return NEU_ERR_EINTERNAL;
+1
View File
@@ -184,6 +184,7 @@ static int azure_parse_config(neu_plugin_t *plugin, const char *setting,
config->cert = cert.v.val_str;
config->key = key.v.val_str;
config->upload_drv_state = false;
config->enable_topic = true;
plog_notice(plugin, "config MQTT version : %d", config->version);
plog_notice(plugin, "config client-id : %s", config->client_id);
+14 -4
View File
@@ -109,6 +109,16 @@
"default": true,
"valid": {}
},
"enable_topic": {
"name": "Enable driver topic",
"name_zh": "启动驱动相关主题",
"description": "Subscription driver related topics, including read/req, write/req and action/req.",
"description_zh": "是否订阅驱动相关主题,包括 read/req、write/req 和 action/req。",
"attribute": "optional",
"type": "bool",
"default": true,
"valid": {}
},
"write-req-topic": {
"name": "Write Request Topic",
"name_zh": "写请求主题",
@@ -116,7 +126,7 @@
"description_zh": "接收点位写入请求的 MQTT 主题。",
"attribute": "required",
"type": "string",
"default": "/neuron/${random_str}/write/req",
"default": "neuron/${random_str}/write/req",
"valid": {
"length": 255
}
@@ -128,7 +138,7 @@
"description_zh": "发送点位写入响应的 MQTT 主题。",
"attribute": "required",
"type": "string",
"default": "/neuron/${random_str}/write/resp",
"default": "neuron/${random_str}/write/resp",
"valid": {
"length": 255
}
@@ -139,7 +149,7 @@
"description": "MQTT topic prefix for driver function.",
"attribute": "required",
"type": "string",
"default": "/neuron/${random_str}",
"default": "neuron/${random_str}",
"valid": {
"length": 128
}
@@ -163,7 +173,7 @@
"field": "upload_drv_state",
"value": true
},
"default": "/neuron/${random_str}/state/update",
"default": "neuron/${random_str}/state/update",
"valid": {
"length": 255
}
+20 -7
View File
@@ -291,13 +291,19 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
.t = NEU_JSON_INT };
neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR };
neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT };
neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR };
neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR };
neu_json_elem_t ssl = { .name = "ssl", .t = NEU_JSON_BOOL };
neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR };
neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR };
neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR };
neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR };
neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR };
neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR };
neu_json_elem_t ssl = { .name = "ssl", .t = NEU_JSON_BOOL };
neu_json_elem_t ca = { .name = "ca", .t = NEU_JSON_STR };
neu_json_elem_t cert = { .name = "cert", .t = NEU_JSON_STR };
neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR };
neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR };
neu_json_elem_t enable_topic = {
.name = "enable_topic",
.t = NEU_JSON_BOOL,
.v.val_bool = true,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
};
neu_json_elem_t upload_drv_state = {
.name = "upload_drv_state",
.t = NEU_JSON_BOOL,
@@ -374,6 +380,11 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
free(schema.v.val_str);
}
ret = neu_parse_param(setting, &err_param, 1, &enable_topic);
if (0 != ret) {
enable_topic.v.val_bool = true; // default to true
}
ret = neu_parse_param(setting, &err_param, 1, &driver_topic_prefix);
if (0 != ret) {
plog_warn(plugin, "parsing driver topic prefix fail, key: `%s`",
@@ -456,6 +467,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
config->client_id = client_id.v.val_str;
config->qos = qos.v.val_int;
config->format = format.v.val_int;
config->enable_topic = enable_topic.v.val_bool;
config->write_req_topic = write_req_topic.v.val_str;
config->write_resp_topic = write_resp_topic.v.val_str;
config->cache = offline_cache.v.val_bool;
@@ -483,6 +495,7 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
plog_notice(plugin, "config qos : %d", config->qos);
plog_notice(plugin, "config format : %s",
mqtt_upload_format_str(config->format));
plog_notice(plugin, "config enable-topic : %d", config->enable_topic);
plog_notice(plugin, "config write-req-topic : %s", config->write_req_topic);
plog_notice(plugin, "config write-resp-topic: %s",
config->write_resp_topic);
+7 -7
View File
@@ -94,13 +94,14 @@ typedef struct {
} mqtt_driver_topic_t;
typedef struct {
neu_mqtt_version_e version; // mqtt version
char * client_id; // client id
neu_mqtt_qos_e qos; // message QoS
mqtt_upload_format_e format; // upload format
char * write_req_topic; // write request topic
char * write_resp_topic; // write response topic
neu_mqtt_version_e version; // mqtt version
char * client_id; // client id
neu_mqtt_qos_e qos; // message QoS
mqtt_upload_format_e format; // upload format
bool enable_topic; // default true
char * write_req_topic; // write request topic
char * write_resp_topic; // write response topic
char * driver_topic_prefix; // driver topic prefix
mqtt_driver_topic_t driver_topic;
@@ -123,7 +124,6 @@ typedef struct {
char * keypass; // client key password
// remove in 2.6, keep it here
// for backward compatibility
size_t n_schema_vt;
mqtt_schema_vt_t *schema_vts;
} mqtt_config_t;
+47 -43
View File
@@ -43,15 +43,28 @@ static void to_traceparent(uint8_t *trace_id, char *span_id, char *out)
}
static int tag_values_to_json(UT_array *tags, mqtt_static_vt_t *s_tags,
size_t n_s_tags, neu_json_read_resp_t *json)
size_t n_s_tags, neu_json_read_resp_t *json,
bool filter_error)
{
int index = 0;
int index = 0;
int n_valid = 0;
if (0 == utarray_len(tags)) {
if (filter_error) {
utarray_foreach(tags, neu_resp_tag_value_meta_t *, tag_value)
{
if (tag_value->value.type != NEU_TYPE_ERROR) {
n_valid += 1;
}
}
} else {
n_valid = utarray_len(tags);
}
if (n_valid == 0) {
return 0;
}
json->n_tag = utarray_len(tags) + n_s_tags;
json->n_tag = n_valid + n_s_tags;
json->tags = (neu_json_read_resp_tag_t *) calloc(
json->n_tag, sizeof(neu_json_read_resp_tag_t));
if (NULL == json->tags) {
@@ -60,8 +73,15 @@ static int tag_values_to_json(UT_array *tags, mqtt_static_vt_t *s_tags,
utarray_foreach(tags, neu_resp_tag_value_meta_t *, tag_value)
{
neu_tag_value_to_json(tag_value, &json->tags[index]);
index += 1;
if (filter_error) {
if (tag_value->value.type != NEU_TYPE_ERROR) {
neu_tag_value_to_json(tag_value, &json->tags[index]);
index += 1;
}
} else {
neu_tag_value_to_json(tag_value, &json->tags[index]);
index += 1;
}
}
if (s_tags != NULL) {
@@ -77,27 +97,6 @@ static int tag_values_to_json(UT_array *tags, mqtt_static_vt_t *s_tags,
return 0;
}
void filter_error_tags(neu_reqresp_trans_data_t *data)
{
if (!data || !data->tags) {
return;
}
UT_array *filtered_tags;
utarray_new(filtered_tags, neu_resp_tag_value_meta_icd());
neu_resp_tag_value_meta_t *tag_ptr = NULL;
while ((tag_ptr = (neu_resp_tag_value_meta_t *) utarray_next(data->tags,
tag_ptr))) {
if (tag_ptr->value.type != NEU_TYPE_ERROR) {
utarray_push_back(filtered_tags, tag_ptr);
}
}
utarray_free(data->tags);
data->tags = filtered_tags;
}
char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
mqtt_upload_format_e format, mqtt_schema_vt_t *vts,
size_t n_vts, mqtt_static_vt_t *s_tags,
@@ -109,29 +108,32 @@ char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
.timestamp = global_timestamp };
neu_json_read_resp_t json = { 0 };
if (!plugin->config.upload_err && skip != NULL) {
filter_error_tags(data);
if (utarray_len(data->tags) == 0) {
*skip = true;
return NULL;
}
}
if (format == MQTT_UPLOAD_FORMAT_CUSTOM) {
if (0 != tag_values_to_json(data->tags, NULL, 0, &json)) {
if (0 !=
tag_values_to_json(data->tags, NULL, 0, &json,
!plugin->config.upload_err)) {
plog_error(plugin, "tag_values_to_json fail");
return NULL;
}
} else {
if (0 != tag_values_to_json(data->tags, s_tags, n_s_tags, &json)) {
if (0 !=
tag_values_to_json(data->tags, s_tags, n_s_tags, &json,
!plugin->config.upload_err)) {
plog_error(plugin, "tag_values_to_json fail");
return NULL;
}
}
int ret;
if (json.n_tag == 0) {
plog_warn(plugin, "driver:%s group:%s, no valid tags", data->driver,
data->group);
if (skip != NULL) {
*skip = true;
}
return NULL;
}
int ret;
switch (format) {
case MQTT_UPLOAD_FORMAT_VALUES:
neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp1, &header,
@@ -186,7 +188,7 @@ static char *generate_read_resp_json(neu_plugin_t * plugin,
char * json_str = NULL;
neu_json_read_resp_t json = { 0 };
if (0 != tag_values_to_json(data->tags, NULL, 0, &json)) {
if (0 != tag_values_to_json(data->tags, NULL, 0, &json, false)) {
plog_error(plugin, "tag_values_to_json fail");
return NULL;
}
@@ -1184,7 +1186,7 @@ int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
break;
}
for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
for (int i = 0; i < tag_value->n_meta; i++) {
if (strlen(tag_value->metas[i].name) > 0) {
if (strncmp(tag_value->metas[i].name, "q", 1) == 0) {
tag->has_q = true;
@@ -1414,7 +1416,7 @@ int handle_trans_data(neu_plugin_t * plugin,
break;
}
for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
for (int i = 0; i < tag_value->n_meta; i++) {
if (strlen(tag_value->metas[i].name) > 0) {
if (strncmp(tag_value->metas[i].name, "q", 1) == 0) {
tag->has_q = true;
@@ -1489,7 +1491,9 @@ int handle_trans_data(neu_plugin_t * plugin,
plugin, trans_data, plugin->config.format,
plugin->config.schema_vts, plugin->config.n_schema_vt,
static_tags, n_satic_tag, &skip_none);
size = strlen(json_str);
if (json_str != NULL) {
size = strlen(json_str);
}
}
if (n_satic_tag > 0) {
mqtt_static_free(static_tags, n_satic_tag);
+24 -42
View File
@@ -62,7 +62,7 @@ static int start_hearbeat_timer(neu_plugin_t *plugin, uint64_t interval)
}
if (NULL == plugin->events) {
plugin->events = neu_event_new();
plugin->events = neu_event_new(plugin->common.name);
if (NULL == plugin->events) {
plog_error(plugin, "neu_event_new fail");
return NEU_ERR_EINTERNAL;
@@ -265,13 +265,13 @@ static int create_topic(neu_plugin_t *plugin)
return 0;
}
neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
neu_asprintf(&plugin->read_req_topic, "neuron/%s/read/req",
plugin->common.name);
if (NULL == plugin->read_req_topic) {
return -1;
}
neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
neu_asprintf(&plugin->read_resp_topic, "neuron/%s/read/resp",
plugin->common.name);
if (NULL == plugin->read_resp_topic) {
free(plugin->read_req_topic);
@@ -387,7 +387,6 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
int rv = 0;
const char * plugin_name = neu_plugin_module.module_name;
mqtt_config_t config = { 0 };
bool started = false;
rv = plugin->parse_config(plugin, setting, &config);
if (0 != rv) {
@@ -395,33 +394,11 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
return NEU_ERR_NODE_SETTING_INVALID;
}
if (NULL == plugin->client) {
plugin->client = neu_mqtt_client_new(config.version);
if (NULL == plugin->client) {
plog_error(plugin, "neu_mqtt_client_new fail");
rv = NEU_ERR_EINTERNAL;
goto error;
}
} else if (neu_mqtt_client_is_open(plugin->client)) {
started = true;
plugin->unsubscribe(plugin, &plugin->config);
rv = neu_mqtt_client_close(plugin->client);
if (0 != rv) {
plog_error(plugin, "neu_mqtt_client_close fail");
rv = NEU_ERR_EINTERNAL;
goto error;
}
if (neu_mqtt_client_check_version_change(plugin->client,
config.version)) {
neu_mqtt_client_free(plugin->client);
plugin->client = neu_mqtt_client_new(config.version);
}
} else if (neu_mqtt_client_check_version_change(plugin->client,
config.version)) {
// plugin stopped and version changed
if (plugin->client != NULL) {
neu_mqtt_client_close(plugin->client);
neu_mqtt_client_free(plugin->client);
plugin->client = neu_mqtt_client_new(config.version);
}
plugin->client = neu_mqtt_client_new(config.version);
rv = config_mqtt_client(plugin, plugin->client, &config);
if (0 != rv) {
@@ -429,17 +406,17 @@ int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
goto error;
}
if (started) {
if (0 != neu_mqtt_client_open(plugin->client)) {
plog_error(plugin, "neu_mqtt_client_open fail");
rv = NEU_ERR_MQTT_CONNECT_FAILURE;
goto error;
}
if (0 != start_hearbeat_timer(plugin, config.heartbeat_interval)) {
plog_error(plugin, "start hearbeat_timer failed");
rv = NEU_ERR_EINTERNAL;
goto error;
}
if (0 != neu_mqtt_client_open(plugin->client)) {
plog_error(plugin, "neu_mqtt_client_open fail");
rv = NEU_ERR_MQTT_CONNECT_FAILURE;
goto error;
}
if (0 != start_hearbeat_timer(plugin, config.heartbeat_interval)) {
plog_error(plugin, "start hearbeat_timer failed");
rv = NEU_ERR_EINTERNAL;
goto error;
}
if (config.enable_topic) {
if (0 != (rv = plugin->subscribe(plugin, &config))) {
goto error;
}
@@ -483,7 +460,10 @@ int mqtt_plugin_start(neu_plugin_t *plugin)
goto end;
}
rv = plugin->subscribe(plugin, &plugin->config);
rv = 0;
if (plugin->config.enable_topic) {
rv = plugin->subscribe(plugin, &plugin->config);
}
end:
if (0 == rv) {
@@ -499,7 +479,9 @@ end:
int mqtt_plugin_stop(neu_plugin_t *plugin)
{
if (plugin->client) {
plugin->unsubscribe(plugin, &plugin->config);
if (plugin->config.enable_topic) {
plugin->unsubscribe(plugin, &plugin->config);
}
neu_mqtt_client_close(plugin->client);
plog_notice(plugin, "mqtt client closed");
}
+73 -20
View File
@@ -41,27 +41,8 @@ void handle_datalayers_get_groups(nng_aio *aio)
NEU_VALIDATE_JWT(aio);
// required parameter
snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
// optional parameter
ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver));
if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
// optional parameter
ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group));
if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
@@ -111,7 +92,22 @@ void handle_datalayers_get_tags(nng_aio *aio)
NEU_VALIDATE_JWT(aio);
// required parameter
ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver));
if (ret <= 0 || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group));
if (ret <= 0 || (size_t) ret == sizeof(cmd.group)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
ret = neu_plugin_op(plugin, header, &cmd);
@@ -137,6 +133,63 @@ void handle_datalayers_get_tags_resp(nng_aio * aio,
neu_http_ok(aio, result);
free(result);
if (tags) {
utarray_foreach(tags->infos, neu_resp_get_sub_driver_tags_info_t *,
info)
{
utarray_free(info->tags);
}
utarray_free(tags->infos);
}
}
void handle_datalayers_get_tag(nng_aio *aio)
{
int ret = 0;
neu_plugin_t * plugin = neu_rest_get_plugin();
neu_req_get_sub_driver_tags_t cmd = { 0 };
neu_reqresp_head_t header = {
.ctx = aio,
.type = NEU_REQ_GET_DATALAYERS_TAG,
.otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM,
};
NEU_VALIDATE_JWT(aio);
ret = neu_http_get_param_str(aio, "tag", cmd.tag, sizeof(cmd.tag));
if (ret <= 0 || (size_t) ret == sizeof(cmd.tag)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
}
}
void handle_datalayers_get_tag_resp(nng_aio * aio,
neu_resp_get_sub_driver_tags_t *tags)
{
char *result = NULL;
int ret = neu_json_encode_by_fn(
tags, neu_json_encode_datalayers_get_tag_resp, &result);
if (ret != 0 || result == NULL) {
neu_http_response(
aio, 500, "{\"error\":4200,\"reason\":\"failed to encode tags\"}");
return;
}
neu_http_ok(aio, result);
free(result);
if (tags) {
utarray_foreach(tags->infos, neu_resp_get_sub_driver_tags_info_t *,
+4
View File
@@ -32,4 +32,8 @@ void handle_datalayers_get_tags(nng_aio *aio);
void handle_datalayers_get_tags_resp(nng_aio * aio,
neu_resp_get_sub_driver_tags_t *tags);
void handle_datalayers_get_tag(nng_aio *aio);
void handle_datalayers_get_tag_resp(nng_aio * aio,
neu_resp_get_sub_driver_tags_t *tags);
#endif
+115 -2
View File
@@ -406,10 +406,16 @@ void handle_grp_subscribes(nng_aio *aio)
NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_subscribe_groups_req_t,
neu_json_decode_subscribe_groups_req, {
int ret = send_subscribe_groups(aio, req);
if (0 != ret) {
int ret = 0;
if (req->n_group == 0) {
NEU_JSON_RESPONSE_ERROR(
ret, { neu_http_response(aio, ret, result_error); });
} else {
ret = send_subscribe_groups(aio, req);
if (0 != ret) {
NEU_JSON_RESPONSE_ERROR(
ret, { neu_http_response(aio, ret, result_error); });
}
}
})
}
@@ -488,4 +494,111 @@ void handle_grp_get_subscribe_resp(nng_aio * aio,
free(result);
free(sub_grp_configs.groups);
utarray_free(groups->groups);
}
void handle_grp_get_subscribes(nng_aio *aio)
{
int ret = 0;
neu_plugin_t * plugin = neu_rest_get_plugin();
neu_req_get_driver_subscribe_group_t cmd = { 0 };
neu_reqresp_head_t header = {
.ctx = aio,
.type = NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP,
.otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM,
};
NEU_VALIDATE_JWT(aio);
// required parameter
ret = neu_http_get_param_str(aio, "app", cmd.app, sizeof(cmd.app));
if (ret <= 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
// optional parameter
ret = neu_http_get_param_str(aio, "name", cmd.name, sizeof(cmd.name));
if (-1 == ret || (size_t) ret == sizeof(cmd.name)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
}
}
void handle_grp_get_subscribes_resp(nng_aio * aio,
neu_resp_get_subscribe_group_t *drivers)
{
char * result = NULL;
neu_json_get_driver_subscribe_resp_t drivers_resp = { 0 };
utarray_foreach(drivers->groups, neu_resp_driver_subscribe_info_t *, group)
{
// int index = utarray_eltidx(drivers->groups, group);
bool found = false;
for (int k = 0; k < drivers_resp.n_driver; k++) {
if (strcmp(drivers_resp.drivers[k].driver, group->driver) == 0) {
found = true;
drivers_resp.drivers[k].n_group++;
drivers_resp.drivers[k].groups = realloc(
drivers_resp.drivers[k].groups,
drivers_resp.drivers[k].n_group *
sizeof(neu_json_get_driver_subscribe_resp_group_t));
drivers_resp.drivers[k]
.groups[drivers_resp.drivers[k].n_group - 1]
.group = strdup(group->group);
drivers_resp.drivers[k]
.groups[drivers_resp.drivers[k].n_group - 1]
.subscribed = group->subscribed;
break;
}
}
if (!found) {
drivers_resp.n_driver++;
drivers_resp.drivers = realloc(
drivers_resp.drivers,
drivers_resp.n_driver *
sizeof(neu_json_get_driver_subscribe_resp_driver_t));
drivers_resp.drivers[drivers_resp.n_driver - 1].driver =
strdup(group->driver);
drivers_resp.drivers[drivers_resp.n_driver - 1].n_group = 1;
drivers_resp.drivers[drivers_resp.n_driver - 1].groups =
calloc(1, sizeof(neu_json_get_driver_subscribe_resp_group_t));
drivers_resp.drivers[drivers_resp.n_driver - 1].groups[0].group =
strdup(group->group);
drivers_resp.drivers[drivers_resp.n_driver - 1]
.groups[0]
.subscribed = group->subscribed;
}
}
neu_json_encode_by_fn(&drivers_resp,
neu_json_encode_get_driver_subscribe_resp, &result);
neu_http_ok(aio, result);
free(result);
for (int i = 0; i < drivers_resp.n_driver; i++) {
free(drivers_resp.drivers[i].driver);
for (int j = 0; j < drivers_resp.drivers[i].n_group; j++) {
free(drivers_resp.drivers[i].groups[j].group);
}
free(drivers_resp.drivers[i].groups);
}
free(drivers_resp.drivers);
utarray_free(drivers->groups);
}
+4
View File
@@ -39,4 +39,8 @@ void handle_grp_get_subscribe(nng_aio *aio);
void handle_grp_get_subscribe_resp(nng_aio * aio,
neu_resp_get_subscribe_group_t *groups);
void handle_grp_get_subscribes(nng_aio *aio);
void handle_grp_get_subscribes_resp(nng_aio * aio,
neu_resp_get_subscribe_group_t *groups);
#endif
+15
View File
@@ -155,6 +155,9 @@ static struct neu_http_handler cors_handler[] = {
{
.url = "/api/v2/datalayers/tags",
},
{
.url = "/api/v2/datalayers/tag",
},
};
static struct neu_http_handler rest_handlers[] = {
@@ -357,6 +360,12 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/subscribe",
.value.handler = handle_grp_get_subscribe,
},
{
.method = NEU_HTTP_METHOD_GET,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/subscribes",
.value.handler = handle_grp_get_subscribes,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
@@ -513,6 +522,12 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/datalayers/tags",
.value.handler = handle_datalayers_get_tags,
},
{
.method = NEU_HTTP_METHOD_GET,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/datalayers/tag",
.value.handler = handle_datalayers_get_tag,
},
};
void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)
+14
View File
@@ -298,6 +298,20 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_DATALAYERS_TAG:
handle_datalayers_get_tag_resp(header->ctx,
(neu_resp_get_sub_driver_tags_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP:
handle_grp_get_subscribes_resp(header->ctx,
(neu_resp_get_subscribe_group_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_NODE_SETTING:
handle_get_node_setting_resp(header->ctx,
(neu_resp_get_node_setting_t *) data);
+1 -1
View File
@@ -147,7 +147,7 @@ int main(int argc, char *argv[])
.params.tcp_server.stop_listen = stop_listen,
};
events = neu_event_new();
events = neu_event_new("modbus_simulator");
conn = neu_conn_new(&param, NULL, connected, disconnected);
signal(SIGINT, sig_handler);
+1 -1
View File
@@ -267,7 +267,7 @@ int main(int argc, char *argv[])
.params.tty_client.timeout = 3000,
};
events = neu_event_new();
events = neu_event_new("modbus_tty_simulator");
conn = neu_conn_new(&param, NULL, connected, disconnected);
neu_conn_start(conn);
+16 -3
View File
@@ -123,8 +123,19 @@ static void *adapter_consumer(void *arg)
nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
adapter->name, header->sender, header->ctx,
neu_reqresp_type_string(header->type), n);
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
if (adapter->state == NEU_NODE_RUNNING_STATE_RUNNING) {
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
} else {
void *ctx = ((neu_reqresp_trans_data_t *) &header[1])->trace_ctx;
if (neu_otel_data_is_started() && ctx) {
neu_otel_trace_ctx trace = neu_otel_find_trace(ctx);
if (trace) {
neu_otel_trace_reduce_expected_span_num(trace, 1);
}
}
}
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
neu_msg_free(msg);
}
@@ -191,7 +202,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
}
adapter->name = strdup(info->name);
adapter->events = neu_event_new();
adapter->events = neu_event_new("adapter");
adapter->state = NEU_NODE_RUNNING_STATE_INIT;
adapter->handle = info->handle;
adapter->cb_funs.command = callback_funs.command;
@@ -755,8 +766,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_RESP_GET_NODE_SETTING:
case NEU_REQ_UPDATE_GROUP:
case NEU_RESP_GET_SUBSCRIBE_GROUP:
case NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP:
case NEU_RESP_GET_DATALAYERS_GROUPS:
case NEU_RESP_GET_DATALAYERS_TAGS:
case NEU_RESP_GET_DATALAYERS_TAG:
case NEU_RESP_ADD_TAG:
case NEU_RESP_ADD_GTAG:
case NEU_RESP_UPDATE_TAG:
+43 -24
View File
@@ -45,7 +45,8 @@ struct elem {
neu_dvalue_t value;
neu_dvalue_t value_old;
neu_tag_meta_t metas[NEU_TAG_META_SIZE];
neu_tag_meta_t *metas;
int n_meta;
tkey_t key;
UT_hash_handle hh;
@@ -115,6 +116,13 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache)
free(elem->value.value.bools.bools);
}
if (elem->metas != NULL) {
for (int i = 0; i < elem->n_meta; i++) {
neu_free_dvalue(&elem->metas[i].value);
}
free(elem->metas);
}
free(elem);
}
@@ -764,9 +772,16 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
}
elem->value.type = value.type;
memset(elem->metas, 0, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE);
for (int i = 0; i < n_meta; i++) {
memcpy(&elem->metas[i], &metas[i], sizeof(neu_tag_meta_t));
if (metas != NULL && n_meta != 0) {
if (elem->metas != NULL) {
free(elem->metas);
}
elem->metas = calloc(n_meta, sizeof(neu_tag_meta_t));
for (int i = 0; i < n_meta; i++) {
memcpy(&elem->metas[i], &metas[i], sizeof(neu_tag_meta_t));
}
elem->n_meta = n_meta;
}
}
@@ -784,7 +799,7 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
const char *tag, neu_driver_cache_value_t *value,
neu_tag_meta_t *metas, int n_meta)
neu_tag_meta_t **metas, int *n_meta)
{
struct elem *elem = NULL;
int ret = -1;
@@ -798,8 +813,14 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
value->value.type = elem->value.type;
value->value.precision = elem->value.precision;
assert(n_meta <= NEU_TAG_META_SIZE);
memcpy(metas, elem->metas, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE);
// assert(n_meta <= NEU_TAG_META_SIZE);
if (elem->metas) {
*metas = calloc(elem->n_meta, sizeof(neu_tag_meta_t));
memcpy(*metas, elem->metas, sizeof(neu_tag_meta_t) * elem->n_meta);
} else {
*metas = NULL;
}
*n_meta = elem->n_meta;
switch (elem->value.type) {
case NEU_TYPE_INT8:
@@ -941,13 +962,6 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
break;
}
for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
if (strlen(elem->metas[i].name) > 0) {
memcpy(&value->metas[i], &elem->metas[i],
sizeof(neu_tag_meta_t));
}
}
ret = 0;
}
@@ -959,7 +973,7 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
const char *group, const char *tag,
neu_driver_cache_value_t *value,
neu_tag_meta_t *metas, int n_meta)
neu_tag_meta_t **metas, int *n_meta)
{
struct elem *elem = NULL;
int ret = -1;
@@ -973,8 +987,14 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
value->value.type = elem->value.type;
value->value.precision = elem->value.precision;
assert(n_meta <= NEU_TAG_META_SIZE);
memcpy(metas, elem->metas, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE);
// assert(n_meta <= NEU_TAG_META_SIZE);
if (elem->metas) {
*metas = calloc(elem->n_meta, sizeof(neu_tag_meta_t));
memcpy(*metas, elem->metas, sizeof(neu_tag_meta_t) * elem->n_meta);
} else {
*metas = NULL;
}
*n_meta = elem->n_meta;
switch (elem->value.type) {
case NEU_TYPE_INT8:
@@ -1116,13 +1136,6 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
break;
}
for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
if (strlen(elem->metas[i].name) > 0) {
memcpy(&value->metas[i], &elem->metas[i],
sizeof(neu_tag_meta_t));
}
}
if (elem->value.type != NEU_TYPE_ERROR) {
elem->changed = false;
}
@@ -1165,6 +1178,12 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
elem->value.type < NEU_TYPE_ARRAY_STRING) {
free(elem->value.value.bools.bools);
}
if (elem->metas != NULL) {
for (int i = 0; i < elem->n_meta; i++) {
neu_free_dvalue(&elem->metas[i].value);
}
free(elem->metas);
}
free(elem);
}
+4 -5
View File
@@ -50,17 +50,16 @@ void neu_driver_cache_update_trace(neu_driver_cache_t *cache, const char *group,
void *neu_driver_cache_get_trace(neu_driver_cache_t *cache, const char *group);
typedef struct {
neu_dvalue_t value;
int64_t timestamp;
neu_tag_meta_t metas[NEU_TAG_META_SIZE];
neu_dvalue_t value;
int64_t timestamp;
} neu_driver_cache_value_t;
int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
const char *tag, neu_driver_cache_value_t *value,
neu_tag_meta_t *metas, int n_meta);
neu_tag_meta_t **metas, int *n_meta);
int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
const char *group, const char *tag,
neu_driver_cache_value_t *value,
neu_tag_meta_t *metas, int n_meta);
neu_tag_meta_t **metas, int *n_meta);
#endif
+62 -29
View File
@@ -486,6 +486,13 @@ static void update_im(neu_adapter_t *adapter, const char *group,
} else {
neu_free_dvalue(&tag_value->value);
}
if (tag_value->metas != NULL) {
for (int i = 0; i < tag_value->n_meta; i++) {
neu_free_dvalue(&tag_value->metas[i].value);
}
free(tag_value->metas);
}
}
utarray_free(data->tags);
free(data->group);
@@ -537,13 +544,13 @@ neu_adapter_driver_t *neu_adapter_driver_create()
{
neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t));
driver->cache = neu_driver_cache_new();
driver->driver_events = neu_event_new();
driver->adapter.cb_funs.driver.update = update;
driver->adapter.cb_funs.driver.write_response = write_response;
driver->adapter.cb_funs.driver.write_responses = write_responses;
driver->adapter.cb_funs.driver.directory_response = directory_response;
driver->adapter.cb_funs.driver.fup_open_response = fup_open_response;
driver->cache = neu_driver_cache_new();
driver->driver_events = neu_event_new("driver");
driver->adapter.cb_funs.driver.update = update;
driver->adapter.cb_funs.driver.write_response = write_response;
driver->adapter.cb_funs.driver.write_responses = write_responses;
driver->adapter.cb_funs.driver.directory_response = directory_response;
driver->adapter.cb_funs.driver.fup_open_response = fup_open_response;
driver->adapter.cb_funs.driver.fdown_open_response = fdown_open_response;
driver->adapter.cb_funs.driver.fup_data_response = fup_data_response;
driver->adapter.cb_funs.driver.update_im = update_im;
@@ -947,9 +954,17 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
case NEU_TYPE_LWORD:
switch (tag->option.value64.endian) {
case NEU_DATATAG_ENDIAN_B64:
case NEU_DATATAG_ENDIAN_BB64:
value->value.u64 = neu_ntohll(value->value.u64);
break;
case NEU_DATATAG_ENDIAN_LB64:
value->value.u64 = neu_htonlb(value->value.u64);
break;
case NEU_DATATAG_ENDIAN_BL64:
value->value.u64 = neu_htonbl(value->value.u64);
break;
case NEU_DATATAG_ENDIAN_L64:
case NEU_DATATAG_ENDIAN_LL64:
default:
break;
}
@@ -2238,6 +2253,7 @@ static int report_callback(void *usr_data)
data->driver = strdup(group->driver->adapter.name);
data->group = strdup(group->name);
utarray_new(data->tags, neu_resp_tag_value_meta_icd());
utarray_reserve(data->tags, utarray_len(tags));
void *trace_ctx =
neu_driver_cache_get_trace(group->driver->cache, group->name);
@@ -2289,20 +2305,6 @@ static int report_callback(void *usr_data)
utarray_foreach(group->apps, sub_app_t *, app)
{
neu_adapter_t *sub_app =
neu_node_manager_find(g_manager->node_manager, app->app);
if (sub_app != NULL) {
if (sub_app->state != NEU_NODE_RUNNING_STATE_RUNNING) {
neu_trans_data_free(data);
continue;
}
} else {
nlog_error("sub app %s not exist", app->app);
neu_trans_data_free(data);
continue;
}
if (group->driver->adapter.cb_funs.responseto(
&group->driver->adapter, &header, data, app->addr) !=
0) {
@@ -2343,6 +2345,13 @@ static int report_callback(void *usr_data)
} else {
neu_free_dvalue(&tag_value->value);
}
if (tag_value->metas != NULL) {
for (int i = 0; i < tag_value->n_meta; i++) {
neu_free_dvalue(&tag_value->metas[i].value);
}
free(tag_value->metas);
}
}
utarray_free(data->tags);
free(data->group);
@@ -2547,15 +2556,15 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) {
if (neu_driver_cache_meta_get_changed(cache, group, tag->name,
&value, tag_value.metas,
NEU_TAG_META_SIZE) != 0) {
&value, &tag_value.metas,
&tag_value.n_meta) != 0) {
nlog_debug("tag: %s not changed", tag->name);
continue;
}
} else {
if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
tag_value.metas,
NEU_TAG_META_SIZE) != 0) {
&tag_value.metas,
&tag_value.n_meta) != 0) {
strcpy(tag_value.tag, tag->name);
tag_value.value.type = NEU_TYPE_ERROR;
tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY;
@@ -2628,9 +2637,17 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
case NEU_TYPE_UINT64:
switch (tag->option.value64.endian) {
case NEU_DATATAG_ENDIAN_B64:
case NEU_DATATAG_ENDIAN_BB64:
value.value.value.u64 = neu_htonll(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_LB64:
value.value.value.u64 = neu_htonlb(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_BL64:
value.value.value.u64 = neu_htonbl(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_L64:
case NEU_DATATAG_ENDIAN_LL64:
default:
break;
}
@@ -2734,8 +2751,8 @@ static void read_group(int64_t timestamp, int64_t timeout,
tag_value.datatag.bias = tag->bias;
if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
tag_value.metas,
NEU_TAG_META_SIZE) != 0) {
&tag_value.metas,
&tag_value.n_meta) != 0) {
tag_value.value.type = NEU_TYPE_ERROR;
tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY;
@@ -2794,9 +2811,17 @@ static void read_group(int64_t timestamp, int64_t timeout,
case NEU_TYPE_UINT64:
switch (tag->option.value64.endian) {
case NEU_DATATAG_ENDIAN_B64:
case NEU_DATATAG_ENDIAN_BB64:
value.value.value.u64 = neu_htonll(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_LB64:
value.value.value.u64 = neu_htonlb(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_BL64:
value.value.value.u64 = neu_htonbl(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_L64:
case NEU_DATATAG_ENDIAN_LL64:
default:
break;
}
@@ -2914,8 +2939,8 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
}
if (neu_driver_cache_meta_get(cache, group, tag->name, &value,
tag_value.metas,
NEU_TAG_META_SIZE) != 0) {
&tag_value.metas,
&tag_value.n_meta) != 0) {
tag_value.value.type = NEU_TYPE_ERROR;
tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY;
@@ -2974,9 +2999,17 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
case NEU_TYPE_UINT64:
switch (tag->option.value64.endian) {
case NEU_DATATAG_ENDIAN_B64:
case NEU_DATATAG_ENDIAN_BB64:
value.value.value.u64 = neu_htonll(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_LB64:
value.value.value.u64 = neu_htonlb(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_BL64:
value.value.value.u64 = neu_htonbl(value.value.value.u64);
break;
case NEU_DATATAG_ENDIAN_L64:
case NEU_DATATAG_ENDIAN_LL64:
default:
break;
}
+111 -106
View File
@@ -30,112 +30,117 @@ extern "C" {
#include "msg.h"
#define NEU_REQRESP_TYPE_MAP(XX) \
XX(NEU_RESP_ERROR, neu_resp_error_t) \
XX(NEU_REQ_READ_GROUP, neu_req_read_group_t) \
XX(NEU_RESP_READ_GROUP, neu_resp_read_group_t) \
XX(NEU_REQ_READ_GROUP_PAGINATE, neu_req_read_group_paginate_t) \
XX(NEU_RESP_READ_GROUP_PAGINATE, neu_resp_read_group_paginate_t) \
XX(NEU_REQ_TEST_READ_TAG, neu_req_test_read_tag_t) \
XX(NEU_RESP_TEST_READ_TAG, neu_resp_test_read_tag_t) \
XX(NEU_REQ_WRITE_TAG, neu_req_write_tag_t) \
XX(NEU_REQ_WRITE_TAGS, neu_req_write_tags_t) \
XX(NEU_REQ_WRITE_GTAGS, neu_req_write_gtags_t) \
XX(NEU_REQ_SUBSCRIBE_GROUP, neu_req_subscribe_t) \
XX(NEU_REQ_UNSUBSCRIBE_GROUP, neu_req_unsubscribe_t) \
XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP, neu_req_subscribe_t) \
XX(NEU_REQ_SUBSCRIBE_GROUPS, neu_req_subscribe_groups_t) \
XX(NEU_REQ_GET_SUBSCRIBE_GROUP, neu_req_get_subscribe_group_t) \
XX(NEU_RESP_GET_SUBSCRIBE_GROUP, neu_resp_get_subscribe_group_t) \
XX(NEU_REQ_GET_SUB_DRIVER_TAGS, neu_req_get_sub_driver_tags_t) \
XX(NEU_RESP_GET_SUB_DRIVER_TAGS, neu_resp_get_sub_driver_tags_t) \
XX(NEU_REQ_NODE_INIT, neu_req_node_init_t) \
XX(NEU_REQ_NODE_UNINIT, neu_req_node_uninit_t) \
XX(NEU_RESP_NODE_UNINIT, neu_resp_node_uninit_t) \
XX(NEU_REQ_ADD_NODE, neu_req_add_node_t) \
XX(NEU_REQ_UPDATE_NODE, neu_req_update_node_t) \
XX(NEU_REQ_DEL_NODE, neu_req_del_node_t) \
XX(NEU_REQ_GET_NODE, neu_req_get_node_t) \
XX(NEU_RESP_GET_NODE, neu_resp_get_node_t) \
XX(NEU_REQ_NODE_SETTING, neu_req_node_setting_t) \
XX(NEU_REQ_GET_NODE_SETTING, neu_req_get_node_setting_t) \
XX(NEU_RESP_GET_NODE_SETTING, neu_resp_get_node_setting_t) \
XX(NEU_REQ_GET_NODE_STATE, neu_req_get_node_state_t) \
XX(NEU_RESP_GET_NODE_STATE, neu_resp_get_node_state_t) \
XX(NEU_REQ_GET_NODES_STATE, neu_req_get_nodes_state_t) \
XX(NEU_RESP_GET_NODES_STATE, neu_resp_get_nodes_state_t) \
XX(NEU_REQ_NODE_CTL, neu_req_node_ctl_t) \
XX(NEU_REQ_NODE_RENAME, neu_req_node_rename_t) \
XX(NEU_RESP_NODE_RENAME, neu_resp_node_rename_t) \
XX(NEU_REQ_ADD_GROUP, neu_req_add_group_t) \
XX(NEU_REQ_DEL_GROUP, neu_req_del_group_t) \
XX(NEU_REQ_UPDATE_GROUP, neu_req_update_group_t) \
XX(NEU_REQ_UPDATE_DRIVER_GROUP, neu_req_update_group_t) \
XX(NEU_RESP_UPDATE_DRIVER_GROUP, neu_resp_update_group_t) \
XX(NEU_REQ_GET_GROUP, neu_req_get_group_t) \
XX(NEU_RESP_GET_GROUP, neu_resp_get_group_t) \
XX(NEU_REQ_GET_DRIVER_GROUP, neu_req_get_group_t) \
XX(NEU_RESP_GET_DRIVER_GROUP, neu_resp_get_driver_group_t) \
XX(NEU_REQ_ADD_TAG, neu_req_add_tag_t) \
XX(NEU_RESP_ADD_TAG, neu_resp_add_tag_t) \
XX(NEU_REQ_ADD_GTAG, neu_req_add_gtag_t) \
XX(NEU_RESP_ADD_GTAG, neu_resp_add_tag_t) \
XX(NEU_REQ_DEL_TAG, neu_req_del_tag_t) \
XX(NEU_REQ_UPDATE_TAG, neu_req_update_tag_t) \
XX(NEU_RESP_UPDATE_TAG, neu_resp_update_tag_t) \
XX(NEU_REQ_GET_TAG, neu_req_get_tag_t) \
XX(NEU_RESP_GET_TAG, neu_resp_get_tag_t) \
XX(NEU_REQ_ADD_PLUGIN, neu_req_add_plugin_t) \
XX(NEU_REQ_DEL_PLUGIN, neu_req_del_plugin_t) \
XX(NEU_REQ_UPDATE_PLUGIN, neu_req_update_plugin_t) \
XX(NEU_REQ_GET_PLUGIN, neu_req_get_plugin_t) \
XX(NEU_RESP_GET_PLUGIN, neu_resp_get_plugin_t) \
XX(NEU_REQRESP_TRANS_DATA, neu_reqresp_trans_data_t) \
XX(NEU_REQRESP_NODES_STATE, neu_reqresp_nodes_state_t) \
XX(NEU_REQRESP_NODE_DELETED, neu_reqresp_node_deleted_t) \
XX(NEU_REQ_ADD_DRIVERS, neu_req_driver_array_t) \
XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t) \
XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t) \
XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t) \
XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t) \
XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t) \
XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t) \
XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t) \
XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t) \
XX(NEU_REQ_DRIVER_ACTION, neu_req_driver_action_t) \
XX(NEU_RESP_DRIVER_ACTION, neu_resp_driver_action_t) \
XX(NEU_REQ_DRIVER_DIRECTORY, neu_req_driver_directory_t) \
XX(NEU_RESP_DRIVER_DIRECTORY, neu_resp_driver_directory_t) \
XX(NEU_REQ_FUP_OPEN, neu_req_fup_open_t) \
XX(NEU_RESP_FUP_OPEN, neu_resp_fup_open_t) \
XX(NEU_REQ_FUP_DATA, neu_req_fup_data_t) \
XX(NEU_RESP_FUP_DATA, neu_resp_fup_data_t) \
XX(NEU_REQ_FDOWN_OPEN, neu_req_fdown_open_t) \
XX(NEU_RESP_FDOWN_OPEN, neu_resp_fdown_open_t) \
XX(NEU_REQ_FDOWN_DATA, neu_req_fdown_data_t) \
XX(NEU_RESP_FDOWN_DATA, neu_resp_fdown_data_t) \
XX(NEU_REQ_ADD_NODE_EVENT, neu_req_add_node_t) \
XX(NEU_REQ_DEL_NODE_EVENT, neu_req_del_node_t) \
XX(NEU_REQ_NODE_CTL_EVENT, neu_req_node_ctl_t) \
XX(NEU_REQ_NODE_SETTING_EVENT, neu_req_node_setting_t) \
XX(NEU_REQ_ADD_GROUP_EVENT, neu_req_add_group_t) \
XX(NEU_REQ_DEL_GROUP_EVENT, neu_req_del_group_t) \
XX(NEU_REQ_UPDATE_GROUP_EVENT, neu_req_update_group_t) \
XX(NEU_REQ_ADD_TAG_EVENT, neu_req_add_tag_t) \
XX(NEU_REQ_DEL_TAG_EVENT, neu_req_del_tag_t) \
XX(NEU_REQ_UPDATE_TAG_EVENT, neu_req_update_tag_t) \
XX(NEU_REQ_ADD_GTAG_EVENT, neu_req_add_gtag_t) \
XX(NEU_REQ_ADD_PLUGIN_EVENT, neu_req_add_plugin_t) \
XX(NEU_REQ_DEL_PLUGIN_EVENT, neu_req_del_plugin_t) \
XX(NEU_REQ_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t) \
XX(NEU_REQ_UNSUBSCRIBE_GROUP_EVENT, neu_req_unsubscribe_t) \
XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t) \
XX(NEU_REQ_SUBSCRIBE_GROUPS_EVENT, neu_req_subscribe_groups_t) \
XX(NEU_RESP_WRITE_TAGS, neu_resp_write_tags_t) \
XX(NEU_REQ_GET_DATALAYERS_GROUPS, neu_req_get_subscribe_group_t) \
XX(NEU_RESP_GET_DATALAYERS_GROUPS, neu_resp_get_subscribe_group_t) \
XX(NEU_REQ_GET_DATALAYERS_TAGS, neu_req_get_sub_driver_tags_t) \
XX(NEU_RESP_GET_DATALAYERS_TAGS, neu_resp_get_sub_driver_tags_t)
#define NEU_REQRESP_TYPE_MAP(XX) \
XX(NEU_RESP_ERROR, neu_resp_error_t) \
XX(NEU_REQ_READ_GROUP, neu_req_read_group_t) \
XX(NEU_RESP_READ_GROUP, neu_resp_read_group_t) \
XX(NEU_REQ_READ_GROUP_PAGINATE, neu_req_read_group_paginate_t) \
XX(NEU_RESP_READ_GROUP_PAGINATE, neu_resp_read_group_paginate_t) \
XX(NEU_REQ_TEST_READ_TAG, neu_req_test_read_tag_t) \
XX(NEU_RESP_TEST_READ_TAG, neu_resp_test_read_tag_t) \
XX(NEU_REQ_WRITE_TAG, neu_req_write_tag_t) \
XX(NEU_REQ_WRITE_TAGS, neu_req_write_tags_t) \
XX(NEU_REQ_WRITE_GTAGS, neu_req_write_gtags_t) \
XX(NEU_REQ_SUBSCRIBE_GROUP, neu_req_subscribe_t) \
XX(NEU_REQ_UNSUBSCRIBE_GROUP, neu_req_unsubscribe_t) \
XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP, neu_req_subscribe_t) \
XX(NEU_REQ_SUBSCRIBE_GROUPS, neu_req_subscribe_groups_t) \
XX(NEU_REQ_GET_SUBSCRIBE_GROUP, neu_req_get_subscribe_group_t) \
XX(NEU_RESP_GET_SUBSCRIBE_GROUP, neu_resp_get_subscribe_group_t) \
XX(NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP, \
neu_req_get_driver_subscribe_group_t) \
XX(NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP, neu_resp_get_subscribe_group_t) \
XX(NEU_REQ_GET_SUB_DRIVER_TAGS, neu_req_get_sub_driver_tags_t) \
XX(NEU_RESP_GET_SUB_DRIVER_TAGS, neu_resp_get_sub_driver_tags_t) \
XX(NEU_REQ_NODE_INIT, neu_req_node_init_t) \
XX(NEU_REQ_NODE_UNINIT, neu_req_node_uninit_t) \
XX(NEU_RESP_NODE_UNINIT, neu_resp_node_uninit_t) \
XX(NEU_REQ_ADD_NODE, neu_req_add_node_t) \
XX(NEU_REQ_UPDATE_NODE, neu_req_update_node_t) \
XX(NEU_REQ_DEL_NODE, neu_req_del_node_t) \
XX(NEU_REQ_GET_NODE, neu_req_get_node_t) \
XX(NEU_RESP_GET_NODE, neu_resp_get_node_t) \
XX(NEU_REQ_NODE_SETTING, neu_req_node_setting_t) \
XX(NEU_REQ_GET_NODE_SETTING, neu_req_get_node_setting_t) \
XX(NEU_RESP_GET_NODE_SETTING, neu_resp_get_node_setting_t) \
XX(NEU_REQ_GET_NODE_STATE, neu_req_get_node_state_t) \
XX(NEU_RESP_GET_NODE_STATE, neu_resp_get_node_state_t) \
XX(NEU_REQ_GET_NODES_STATE, neu_req_get_nodes_state_t) \
XX(NEU_RESP_GET_NODES_STATE, neu_resp_get_nodes_state_t) \
XX(NEU_REQ_NODE_CTL, neu_req_node_ctl_t) \
XX(NEU_REQ_NODE_RENAME, neu_req_node_rename_t) \
XX(NEU_RESP_NODE_RENAME, neu_resp_node_rename_t) \
XX(NEU_REQ_ADD_GROUP, neu_req_add_group_t) \
XX(NEU_REQ_DEL_GROUP, neu_req_del_group_t) \
XX(NEU_REQ_UPDATE_GROUP, neu_req_update_group_t) \
XX(NEU_REQ_UPDATE_DRIVER_GROUP, neu_req_update_group_t) \
XX(NEU_RESP_UPDATE_DRIVER_GROUP, neu_resp_update_group_t) \
XX(NEU_REQ_GET_GROUP, neu_req_get_group_t) \
XX(NEU_RESP_GET_GROUP, neu_resp_get_group_t) \
XX(NEU_REQ_GET_DRIVER_GROUP, neu_req_get_group_t) \
XX(NEU_RESP_GET_DRIVER_GROUP, neu_resp_get_driver_group_t) \
XX(NEU_REQ_ADD_TAG, neu_req_add_tag_t) \
XX(NEU_RESP_ADD_TAG, neu_resp_add_tag_t) \
XX(NEU_REQ_ADD_GTAG, neu_req_add_gtag_t) \
XX(NEU_RESP_ADD_GTAG, neu_resp_add_tag_t) \
XX(NEU_REQ_DEL_TAG, neu_req_del_tag_t) \
XX(NEU_REQ_UPDATE_TAG, neu_req_update_tag_t) \
XX(NEU_RESP_UPDATE_TAG, neu_resp_update_tag_t) \
XX(NEU_REQ_GET_TAG, neu_req_get_tag_t) \
XX(NEU_RESP_GET_TAG, neu_resp_get_tag_t) \
XX(NEU_REQ_ADD_PLUGIN, neu_req_add_plugin_t) \
XX(NEU_REQ_DEL_PLUGIN, neu_req_del_plugin_t) \
XX(NEU_REQ_UPDATE_PLUGIN, neu_req_update_plugin_t) \
XX(NEU_REQ_GET_PLUGIN, neu_req_get_plugin_t) \
XX(NEU_RESP_GET_PLUGIN, neu_resp_get_plugin_t) \
XX(NEU_REQRESP_TRANS_DATA, neu_reqresp_trans_data_t) \
XX(NEU_REQRESP_NODES_STATE, neu_reqresp_nodes_state_t) \
XX(NEU_REQRESP_NODE_DELETED, neu_reqresp_node_deleted_t) \
XX(NEU_REQ_ADD_DRIVERS, neu_req_driver_array_t) \
XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t) \
XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t) \
XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t) \
XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t) \
XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t) \
XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t) \
XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t) \
XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t) \
XX(NEU_REQ_DRIVER_ACTION, neu_req_driver_action_t) \
XX(NEU_RESP_DRIVER_ACTION, neu_resp_driver_action_t) \
XX(NEU_REQ_DRIVER_DIRECTORY, neu_req_driver_directory_t) \
XX(NEU_RESP_DRIVER_DIRECTORY, neu_resp_driver_directory_t) \
XX(NEU_REQ_FUP_OPEN, neu_req_fup_open_t) \
XX(NEU_RESP_FUP_OPEN, neu_resp_fup_open_t) \
XX(NEU_REQ_FUP_DATA, neu_req_fup_data_t) \
XX(NEU_RESP_FUP_DATA, neu_resp_fup_data_t) \
XX(NEU_REQ_FDOWN_OPEN, neu_req_fdown_open_t) \
XX(NEU_RESP_FDOWN_OPEN, neu_resp_fdown_open_t) \
XX(NEU_REQ_FDOWN_DATA, neu_req_fdown_data_t) \
XX(NEU_RESP_FDOWN_DATA, neu_resp_fdown_data_t) \
XX(NEU_REQ_ADD_NODE_EVENT, neu_req_add_node_t) \
XX(NEU_REQ_DEL_NODE_EVENT, neu_req_del_node_t) \
XX(NEU_REQ_NODE_CTL_EVENT, neu_req_node_ctl_t) \
XX(NEU_REQ_NODE_SETTING_EVENT, neu_req_node_setting_t) \
XX(NEU_REQ_ADD_GROUP_EVENT, neu_req_add_group_t) \
XX(NEU_REQ_DEL_GROUP_EVENT, neu_req_del_group_t) \
XX(NEU_REQ_UPDATE_GROUP_EVENT, neu_req_update_group_t) \
XX(NEU_REQ_ADD_TAG_EVENT, neu_req_add_tag_t) \
XX(NEU_REQ_DEL_TAG_EVENT, neu_req_del_tag_t) \
XX(NEU_REQ_UPDATE_TAG_EVENT, neu_req_update_tag_t) \
XX(NEU_REQ_ADD_GTAG_EVENT, neu_req_add_gtag_t) \
XX(NEU_REQ_ADD_PLUGIN_EVENT, neu_req_add_plugin_t) \
XX(NEU_REQ_DEL_PLUGIN_EVENT, neu_req_del_plugin_t) \
XX(NEU_REQ_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t) \
XX(NEU_REQ_UNSUBSCRIBE_GROUP_EVENT, neu_req_unsubscribe_t) \
XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t) \
XX(NEU_REQ_SUBSCRIBE_GROUPS_EVENT, neu_req_subscribe_groups_t) \
XX(NEU_RESP_WRITE_TAGS, neu_resp_write_tags_t) \
XX(NEU_REQ_GET_DATALAYERS_GROUPS, neu_req_get_subscribe_group_t) \
XX(NEU_RESP_GET_DATALAYERS_GROUPS, neu_resp_get_subscribe_group_t) \
XX(NEU_REQ_GET_DATALAYERS_TAGS, neu_req_get_sub_driver_tags_t) \
XX(NEU_RESP_GET_DATALAYERS_TAGS, neu_resp_get_sub_driver_tags_t) \
XX(NEU_REQ_GET_DATALAYERS_TAG, neu_req_get_sub_driver_tags_t) \
XX(NEU_RESP_GET_DATALAYERS_TAG, neu_resp_get_sub_driver_tags_t)
static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
{
+27 -13
View File
@@ -262,21 +262,35 @@ int neu_datatag_parse_addr_option(const neu_datatag_t * datatag,
case NEU_TYPE_UINT64: {
char *op = find_last_character(datatag->address, '#');
option->value64.endian = NEU_DATATAG_ENDIAN_L64;
option->value64.endian = NEU_DATATAG_ENDIAN_L64;
option->value64.is_default = true;
if (op != NULL) {
char e = 0;
sscanf(op, "#%c", &e);
char e1 = 0;
char e2 = 0;
int n = sscanf(op, "#%c%c", &e1, &e2);
switch (e) {
case 'B':
option->value64.endian = NEU_DATATAG_ENDIAN_B64;
break;
case 'L':
option->value64.endian = NEU_DATATAG_ENDIAN_L64;
break;
default:
option->value64.endian = NEU_DATATAG_ENDIAN_L64;
break;
if (n == 2) {
if (e1 == 'B' && e2 == 'B') {
option->value64.endian = NEU_DATATAG_ENDIAN_BB64;
option->value64.is_default = false;
} else if (e1 == 'B' && e2 == 'L') {
option->value64.endian = NEU_DATATAG_ENDIAN_BL64;
option->value64.is_default = false;
} else if (e1 == 'L' && e2 == 'L') {
option->value64.endian = NEU_DATATAG_ENDIAN_LL64;
option->value64.is_default = false;
} else if (e1 == 'L' && e2 == 'B') {
option->value64.endian = NEU_DATATAG_ENDIAN_LB64;
option->value64.is_default = false;
}
} else if (n == 1) {
if (e1 == 'B') {
option->value64.endian = NEU_DATATAG_ENDIAN_B64;
option->value64.is_default = false;
} else if (e1 == 'L') {
option->value64.endian = NEU_DATATAG_ENDIAN_L64;
option->value64.is_default = false;
}
}
}
+1 -1
View File
@@ -503,7 +503,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
conn->param.log,
"tcp conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
conn->fd, len, ret, strerror(errno), errno);
if (ret == 0 || (ret == -1 && errno != EAGAIN)) {
if (ret == 0 || (conn->connection_ok && ret == -1 && errno != EAGAIN)) {
conn_disconnect(conn);
}
}
+1 -1
View File
@@ -137,7 +137,7 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
conn_eth->ic = &in_conns[i];
in_conns[i].interface = strdup(interface);
in_conns[i].events = neu_event_new();
in_conns[i].events = neu_event_new("eth_conn");
in_conns[i].count += 1;
get_mac(conn_eth, interface);
+1 -1
View File
@@ -907,7 +907,7 @@ static inline int client_start_timer(neu_mqtt_client_t *client)
return 0;
}
events = neu_event_new();
events = neu_event_new("mqtt_client");
if (NULL == events) {
return -1;
}
+91 -1
View File
@@ -98,7 +98,7 @@ neu_manager_t *neu_manager_create()
.type = NEU_EVENT_TIMER_NOBLOCK,
};
manager->events = neu_event_new();
manager->events = neu_event_new("manager");
manager->plugin_manager = neu_plugin_manager_create();
manager->node_manager = neu_node_manager_create();
manager->subscribe_manager = neu_subscribe_manager_create();
@@ -1010,6 +1010,18 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_DRIVER_SUBSCRIBE_GROUP: {
neu_req_get_driver_subscribe_group_t *cmd =
(neu_req_get_driver_subscribe_group_t *) &header[1];
UT_array *groups =
neu_manager_get_driver_groups(manager, cmd->app, cmd->name);
neu_resp_get_subscribe_group_t resp = { .groups = groups };
strcpy(header->receiver, header->sender);
header->type = NEU_RESP_GET_DRIVER_SUBSCRIBE_GROUP;
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_SUB_DRIVER_TAGS: {
neu_req_get_sub_driver_tags_t *cmd =
(neu_req_get_sub_driver_tags_t *) &header[1];
@@ -1056,10 +1068,18 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
(neu_req_get_sub_driver_tags_t *) &header[1];
neu_resp_get_sub_driver_tags_t resp = { 0 };
UT_array *groups = neu_manager_get_sub_group(manager, cmd->app);
bool found = false;
utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd());
utarray_foreach(groups, neu_resp_subscribe_info_t *, info)
{
if (strcmp(info->driver, cmd->driver) != 0 ||
strcmp(info->group, cmd->group) != 0) {
continue;
}
found = true;
neu_resp_get_sub_driver_tags_info_t in = { 0 };
neu_adapter_t * driver =
neu_node_manager_find(manager->node_manager, info->driver);
@@ -1074,12 +1094,82 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
utarray_free(groups);
if (!found) {
utarray_free(resp.infos);
neu_resp_error_t e = { .error = NEU_ERR_GROUP_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
break;
}
strcpy(header->receiver, header->sender);
header->type = NEU_RESP_GET_DATALAYERS_TAGS;
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_DATALAYERS_TAG: {
neu_req_get_sub_driver_tags_t *cmd =
(neu_req_get_sub_driver_tags_t *) &header[1];
neu_resp_get_sub_driver_tags_t resp = { 0 };
UT_array *groups = neu_manager_get_sub_group(manager, cmd->app);
bool found = false;
utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd());
utarray_foreach(groups, neu_resp_subscribe_info_t *, info)
{
neu_resp_get_sub_driver_tags_info_t in = { 0 };
neu_adapter_t * driver =
neu_node_manager_find(manager->node_manager, info->driver);
assert(driver != NULL);
neu_adapter_driver_get_value_tag((neu_adapter_driver_t *) driver,
info->group, &in.tags);
if (in.tags) {
UT_array *filtered_tags = NULL;
utarray_new(filtered_tags, neu_tag_get_icd());
utarray_foreach(in.tags, neu_datatag_t *, tag)
{
if (strstr(tag->name, cmd->tag) != NULL) {
utarray_push_back(filtered_tags, tag);
}
}
utarray_free(in.tags);
if (utarray_len(filtered_tags) > 0) {
strcpy(in.driver, info->driver);
strcpy(in.group, info->group);
in.tags = filtered_tags;
utarray_push_back(resp.infos, &in);
found = true;
} else {
utarray_free(filtered_tags);
}
}
}
utarray_free(groups);
if (!found) {
utarray_free(resp.infos);
neu_resp_error_t e = { .error = NEU_ERR_TAG_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
break;
}
strcpy(header->receiver, header->sender);
header->type = NEU_RESP_GET_DATALAYERS_TAG;
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_NODES_STATE: {
neu_resp_get_nodes_state_t resp = { 0 };
UT_array *states = neu_node_manager_get_state(manager->node_manager);
+49
View File
@@ -380,6 +380,55 @@ UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
return subs;
}
UT_array *neu_manager_get_driver_groups(neu_manager_t *manager, const char *app,
const char *name)
{
UT_icd icd = { sizeof(neu_resp_driver_subscribe_info_t), NULL, NULL, NULL };
UT_array *result = NULL;
UT_array *sub_groups =
neu_subscribe_manager_get(manager->subscribe_manager, app, NULL, NULL);
UT_array *driver_groups = neu_manager_get_driver_group(manager);
utarray_new(result, &icd);
utarray_foreach(driver_groups, neu_resp_driver_group_info_t *, grp)
{
neu_resp_driver_subscribe_info_t info = { 0 };
if (name != NULL && strlen(name) > 0) {
if (strstr(grp->driver, name) != NULL ||
strstr(grp->group, name) != NULL) {
strcpy(info.app, app);
strcpy(info.driver, grp->driver);
strcpy(info.group, grp->group);
info.subscribed = false;
}
} else {
strcpy(info.app, app);
strcpy(info.driver, grp->driver);
strcpy(info.group, grp->group);
info.subscribed = false;
}
if (strlen(info.driver) > 0 && strlen(info.group) > 0) {
utarray_foreach(sub_groups, neu_resp_subscribe_info_t *, sub)
{
if (strcmp(sub->driver, info.driver) == 0 &&
strcmp(sub->group, info.group) == 0) {
info.subscribed = true;
break;
}
}
utarray_push_back(result, &info);
}
}
utarray_free(sub_groups);
utarray_free(driver_groups);
return result;
}
int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
neu_persist_node_info_t *info)
{
+2
View File
@@ -85,6 +85,8 @@ UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
const char * app,
const char * driver,
const char * group);
UT_array *neu_manager_get_driver_groups(neu_manager_t *manager, const char *app,
const char *name);
int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
neu_persist_node_info_t *info);
+13
View File
@@ -18,6 +18,7 @@
**/
#include "errcodes.h"
#include "utils/log.h"
#include "utils/time.h"
#include "adapter/storage.h"
#include "storage.h"
@@ -140,6 +141,8 @@ int manager_load_node(neu_manager_t *manager)
return -1;
}
int i = 0;
utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
{
rv = neu_manager_add_node(manager, node_info->name,
@@ -149,6 +152,10 @@ int manager_load_node(neu_manager_t *manager)
nlog_notice("load adapter %s type:%d, name:%s plugin:%s state:%d",
ok_or_err, node_info->type, node_info->name,
node_info->plugin_name, node_info->state);
i++;
if (i % 10 == 0) {
neu_msleep(10);
}
}
utarray_free(node_infos);
@@ -160,6 +167,8 @@ int manager_load_subscribe(neu_manager_t *manager)
UT_array *nodes =
neu_node_manager_get(manager->node_manager, NEU_NA_TYPE_APP);
int i = 0;
utarray_foreach(nodes, neu_resp_node_info_t *, node)
{
int rv = 0;
@@ -186,6 +195,10 @@ int manager_load_subscribe(neu_manager_t *manager)
info->group_name, app_port,
info->params, info->static_tags);
}
i++;
if (i % 10 == 0) {
neu_msleep(10);
}
}
utarray_free(sub_infos);
+21 -16
View File
@@ -72,6 +72,7 @@ struct neu_events {
int epoll_fd;
pthread_t thread;
bool stop;
char * name;
pthread_mutex_t mtx;
int n_event;
@@ -122,8 +123,8 @@ static void *event_loop(void *arg)
}
if (ret == -1 || events->stop) {
zlog_warn(neuron, "event loop exit, errno: %s(%d), stop: %d",
strerror(errno), errno, events->stop);
zlog_warn(neuron, "event loop(%s) exit, errno: %s(%d), stop: %d",
events->name, strerror(errno), errno, events->stop);
break;
}
@@ -177,17 +178,18 @@ static void *event_loop(void *arg)
return NULL;
};
neu_events_t *neu_event_new(void)
neu_events_t *neu_event_new(const char *name)
{
neu_events_t *events = calloc(1, sizeof(struct neu_events));
events->epoll_fd = epoll_create(1);
nlog_notice("create epoll: %d(%d)", events->epoll_fd, errno);
nlog_notice("create epoll(%s): %d(%d)", name, events->epoll_fd, errno);
assert(events->epoll_fd > 0);
events->stop = false;
events->n_event = 0;
events->name = strdup(name);
pthread_mutex_init(&events->mtx, NULL);
pthread_create(&events->thread, NULL, event_loop, events);
@@ -199,6 +201,7 @@ int neu_event_close(neu_events_t *events)
{
events->stop = true;
close(events->epoll_fd);
free(events->name);
pthread_join(events->thread, NULL);
pthread_mutex_destroy(&events->mtx);
@@ -220,7 +223,8 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t * events,
};
int index = get_free_event(events);
if (index < 0) {
zlog_fatal(neuron, "no free event: %d", events->epoll_fd);
zlog_fatal(neuron, "no free event(%s): %d", events->name,
events->epoll_fd);
}
assert(index >= 0);
@@ -251,18 +255,19 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t * events,
zlog_notice(neuron,
"add timer, second: %" PRId64 ", millisecond: %" PRId64
", timer: %d in epoll %d, "
", timer: %d in epoll(%s) %d, "
"ret: %d, index: %d",
timer.second, timer.millisecond, timer_fd, events->epoll_fd,
ret, index);
timer.second, timer.millisecond, timer_fd, events->name,
events->epoll_fd, ret, index);
return timer_ctx;
}
int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
{
zlog_notice(neuron, "del timer: %d from epoll: %d, index: %d", timer->fd,
events->epoll_fd, timer->event_data->index);
zlog_notice(neuron, "del timer: %d from epoll(%s): %d, index: %d",
timer->fd, events->name, events->epoll_fd,
timer->event_data->index);
timer->stop = true;
epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, timer->fd, NULL);
@@ -281,8 +286,8 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
int ret = 0;
int index = get_free_event(events);
nlog_notice("add io, fd: %d, epoll: %d, index: %d", io.fd, events->epoll_fd,
index);
nlog_notice("add io, fd: %d, epoll(%s): %d, index: %d", io.fd, events->name,
events->epoll_fd, index);
assert(index >= 0);
neu_event_io_t *io_ctx = &events->event_datas[index].ctx.io;
@@ -303,8 +308,8 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, io.fd, &event);
nlog_notice("add io, fd: %d, epoll: %d, ret: %d(%d), index: %d", io.fd,
events->epoll_fd, ret, errno, index);
nlog_notice("add io, fd: %d, epoll(%s): %d, ret: %d(%d), index: %d", io.fd,
events->name, events->epoll_fd, ret, errno, index);
assert(ret == 0);
return io_ctx;
@@ -316,8 +321,8 @@ int neu_event_del_io(neu_events_t *events, neu_event_io_t *io)
return 0;
}
zlog_notice(neuron, "del io: %d from epoll: %d, index: %d", io->fd,
events->epoll_fd, io->event_data->index);
zlog_notice(neuron, "del io: %d from epoll(%s): %d, index: %d", io->fd,
events->name, events->epoll_fd, io->event_data->index);
epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, io->fd, NULL);
free_event(events, io->event_data->index);
+11 -1
View File
@@ -540,6 +540,16 @@ void neu_otel_trace_set_expected_span_num(neu_otel_trace_ctx ctx, uint32_t num)
trace_ctx->expected_span_num = num;
}
void neu_otel_trace_reduce_expected_span_num(neu_otel_trace_ctx ctx,
uint32_t num)
{
trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;
trace_ctx->expected_span_num -= num;
if (trace_ctx->expected_span_num <= 0) {
trace_ctx->final = true;
}
}
uint8_t *neu_otel_get_trace_id(neu_otel_trace_ctx ctx)
{
trace_ctx_t *trace_ctx = (trace_ctx_t *) ctx;
@@ -1103,7 +1113,7 @@ static int otel_timer_cb(void *data)
void neu_otel_start()
{
if (otel_event == NULL) {
otel_event = neu_event_new();
otel_event = neu_event_new("otel_event");
}
if (otel_timer == NULL) {
+108
View File
@@ -163,6 +163,7 @@ int neu_json_encode_datalayers_get_tags_resp(void *json_object_param,
json_object_set_new(tag_obj, "name", json_string(tag->name));
json_object_set_new(tag_obj, "type", json_integer(tag->type));
json_object_set_new(tag_obj, "category", json_integer(3));
json_object_set_new(tag_obj, "leaf", json_true());
json_object_set_new(tag_obj, "children", json_array());
json_array_append_new(tag_array, tag_obj);
@@ -183,6 +184,113 @@ int neu_json_encode_datalayers_get_tags_resp(void *json_object_param,
.v.val_object = final_obj,
} };
int ret = neu_json_encode_field(json_object_param, elems,
NEU_JSON_ELEM_SIZE(elems));
json_decref(final_obj);
return ret;
}
int neu_json_encode_datalayers_get_tag_resp(void *json_object_param,
void *param)
{
neu_resp_get_sub_driver_tags_t *resp =
(neu_resp_get_sub_driver_tags_t *) param;
json_t *node_array = json_array();
for (unsigned int i = 0; i < utarray_len(resp->infos); i++) {
neu_resp_get_sub_driver_tags_info_t *info =
(neu_resp_get_sub_driver_tags_info_t *) utarray_eltptr(resp->infos,
i);
char node_id[37];
char node_uuid_input[128];
snprintf(node_uuid_input, sizeof(node_uuid_input), "%s", info->driver);
generate_fixed_uuid(node_uuid_input, node_id);
json_t *node_obj = NULL;
for (size_t j = 0; j < json_array_size(node_array); j++) {
node_obj = json_array_get(node_array, j);
if (strcmp(json_string_value(json_object_get(node_obj, "id")),
node_id) == 0) {
break;
}
node_obj = NULL;
}
if (!node_obj) {
node_obj = json_object();
json_object_set_new(node_obj, "id", json_string(node_id));
json_object_set_new(node_obj, "name", json_string(info->driver));
json_object_set_new(node_obj, "category", json_integer(1));
json_t *node_children = json_array();
json_object_set_new(node_obj, "children", node_children);
json_array_append_new(node_array, node_obj);
}
json_t *group_obj = NULL;
char group_id[37];
char group_uuid_input[256];
snprintf(group_uuid_input, sizeof(group_uuid_input), "%s-%s",
info->driver, info->group);
generate_fixed_uuid(group_uuid_input, group_id);
json_t *children_array = json_object_get(node_obj, "children");
for (size_t k = 0; k < json_array_size(children_array); k++) {
group_obj = json_array_get(children_array, k);
if (strcmp(json_string_value(json_object_get(group_obj, "id")),
group_id) == 0) {
break;
}
group_obj = NULL;
}
if (!group_obj) {
group_obj = json_object();
json_object_set_new(group_obj, "id", json_string(group_id));
json_object_set_new(group_obj, "name", json_string(info->group));
json_object_set_new(group_obj, "category", json_integer(2));
json_t *group_children = json_array();
json_object_set_new(group_obj, "children", group_children);
json_array_append_new(children_array, group_obj);
}
json_t *tag_array = json_object_get(group_obj, "children");
if (info->tags) {
for (unsigned int j = 0; j < utarray_len(info->tags); j++) {
neu_datatag_t *tag =
(neu_datatag_t *) utarray_eltptr(info->tags, j);
char tag_id[37];
char tag_uuid_input[512];
snprintf(tag_uuid_input, sizeof(tag_uuid_input), "%s-%s-%s",
info->driver, info->group, tag->name);
generate_fixed_uuid(tag_uuid_input, tag_id);
json_t *tag_obj = json_object();
json_object_set_new(tag_obj, "id", json_string(tag_id));
json_object_set_new(tag_obj, "name", json_string(tag->name));
json_object_set_new(tag_obj, "type", json_integer(tag->type));
json_object_set_new(tag_obj, "category", json_integer(3));
json_object_set_new(tag_obj, "leaf", json_true());
json_object_set_new(tag_obj, "children", json_array());
json_array_append_new(tag_array, tag_obj);
}
}
}
json_t *final_obj = json_object();
json_object_set_new(final_obj, "error", json_integer(0));
json_object_set_new(final_obj, "children", node_array);
neu_json_elem_t elems[] = { {
.name = NULL,
.t = NEU_JSON_OBJECT,
.v.val_object = final_obj,
} };
int ret = neu_json_encode_field(json_object_param, elems,
NEU_JSON_ELEM_SIZE(elems));
json_decref(final_obj);
+2
View File
@@ -47,6 +47,8 @@ int neu_json_encode_datalayers_get_groups_resp(void *json_object, void *param);
int neu_json_encode_datalayers_get_tags_resp(void *json_object, void *param);
int neu_json_encode_datalayers_get_tag_resp(void *json_object, void *param);
#ifdef __cplusplus
}
#endif
+44
View File
@@ -405,6 +405,50 @@ int neu_json_encode_get_subscribe_resp(void *object, void *param)
return ret;
}
int neu_json_encode_get_driver_subscribe_resp(void *object, void *param)
{
neu_json_get_driver_subscribe_resp_t *resp =
(neu_json_get_driver_subscribe_resp_t *) param;
void *driver_array = neu_json_array();
for (int i = 0; i < resp->n_driver; i++) {
if (resp->drivers[i].n_group == 0) {
continue;
}
json_t *ob = json_object();
void * group_array = neu_json_array();
json_object_set_new(ob, "name", json_string(resp->drivers[i].driver));
for (int j = 0; j < resp->drivers[i].n_group; j++) {
json_t *group_ob = json_object();
json_object_set_new(group_ob, "name",
json_string(resp->drivers[i].groups[j].group));
json_object_set_new(
group_ob, "subscribed",
json_boolean(resp->drivers[i].groups[j].subscribed));
json_array_append_new(group_array, group_ob);
}
json_object_set_new(ob, "groups", group_array);
json_array_append_new(driver_array, ob);
}
neu_json_elem_t resp_elems[] = { {
.name = "drivers",
.t = NEU_JSON_OBJECT,
.v.val_object = driver_array,
} };
return neu_json_encode_field(object, resp_elems,
NEU_JSON_ELEM_SIZE(resp_elems));
}
static inline int dump_params(void *root, char **const result)
{
return neu_json_dump_key(root, "params", result, false);
+18
View File
@@ -113,6 +113,24 @@ typedef struct {
int neu_json_encode_get_subscribe_resp(void *json_object, void *param);
typedef struct {
char *group;
bool subscribed;
} neu_json_get_driver_subscribe_resp_group_t;
typedef struct {
char * driver;
int n_group;
neu_json_get_driver_subscribe_resp_group_t *groups;
} neu_json_get_driver_subscribe_resp_driver_t;
typedef struct {
int n_driver;
neu_json_get_driver_subscribe_resp_driver_t *drivers;
} neu_json_get_driver_subscribe_resp_t;
int neu_json_encode_get_driver_subscribe_resp(void *json_object, void *param);
typedef struct {
char *group;
char *app;
+4
View File
@@ -534,6 +534,10 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
ele->t = NEU_JSON_ARRAY_BOOL;
break;
}
if (json_is_object(value)) {
ele->t = NEU_JSON_OBJECT;
break;
}
}
} else if (json_is_object(ob)) {
ele->t = NEU_JSON_OBJECT;
+7 -7
View File
@@ -62,12 +62,12 @@ TAGS = [
]
FORMAT_VALUES = 0
FORMAT_TAGS = 1
UPLOAD_TOPIC = f"/neuron/{NODE}/upload"
WRITE_REQ_TOPIC = f"/neuron/{NODE}/write/req"
WRITE_RESP_TOPIC = f"/neuron/{NODE}/write/resp"
READ_REQ_TOPIC = f"/neuron/{NODE}/read/req"
READ_RESP_TOPIC = f"/neuron/{NODE}/read/resp"
UPLOAD_DRV_STATE_TOPIC = f"/neuron/{NODE}/state/update"
UPLOAD_TOPIC = f"neuron/{NODE}/upload"
WRITE_REQ_TOPIC = f"neuron/{NODE}/write/req"
WRITE_RESP_TOPIC = f"neuron/{NODE}/write/resp"
READ_REQ_TOPIC = f"neuron/{NODE}/read/req"
READ_RESP_TOPIC = f"neuron/{NODE}/read/resp"
UPLOAD_DRV_STATE_TOPIC = f"neuron/{NODE}/state/update"
@pytest.fixture(autouse=True, scope="class")
@@ -204,7 +204,7 @@ def conf_upload_driver_state(conf_base):
return {
**conf_base,
"upload_drv_state": True,
"upload_drv_state_topic": f"/neuron/{NODE}/state/update",
"upload_drv_state_topic": f"neuron/{NODE}/state/update",
"upload_drv_state_interval": 1,
}
+1 -1
View File
@@ -1 +1 @@
2.12.0-alpha
2.12.2