diff --git a/ft/common.resource b/ft/common.resource index 97776ac2..dbd53832 100644 --- a/ft/common.resource +++ b/ft/common.resource @@ -19,7 +19,7 @@ ${NODE_UNKNOWN} 0 ${NODE_DRIVER} 1 ${NODE_WEB} 2 ${NODE_MQTT} 3 -${NODE_STREAM_PROCESSOR} 4 +${NODE_DRIVERX} 4 ${NODE_APP} 5 ${NODE_CTL_START} 0 diff --git a/ft/grp_config_api_test.robot b/ft/grp_config_api_test.robot index a5eafa4f..ca7af1ff 100644 --- a/ft/grp_config_api_test.robot +++ b/ft/grp_config_api_test.robot @@ -80,7 +80,7 @@ Update the interval of the test group config, it should return success Should Be Equal As Integers ${config}[interval] 2000 North APP subscribe non-existent group config, it should return failure - ${app_node_id} = Add Node And Return ID ${NODE_APP} ${app_node} ${PLUGIN_MQTT} + ${app_node_id} = Add Node And Return ID ${NODE_MQTT} ${app_node} ${PLUGIN_MQTT} ${driver_node_id} = Add Node And Return ID ${NODE_DRIVER} ${driver_node} ${PLUGIN_MODBUS_TCP} ${res} = Subscribe Group ${driver_node_id} ${app_node_id} grp_config @@ -89,7 +89,7 @@ North APP subscribe non-existent group config, it should return failure Check Error Code ${res} ${ERR_GRPCONFIG_NOT_EXIST} North APP subscribe group config, it should return success - ${app_node_id} = Get Node ID ${NODE_APP} ${app_node} + ${app_node_id} = Get Node ID ${NODE_MQTT} ${app_node} ${driver_node_id} = Get Node ID ${NODE_DRIVER} ${driver_node} ${res} = Add Group Config ${driver_node_id} grp_config 1000 @@ -99,7 +99,7 @@ North APP subscribe group config, it should return success Check Error Code ${res} ${ERR_SUCCESS} Query the subscribers of the group config, it should return all nodes subscribed to this group - ${app_node_id} = Get Node ID ${NODE_APP} ${app_node} + ${app_node_id} = Get Node ID ${NODE_MQTT} ${app_node} ${driver_node_id1} = Get Node ID ${NODE_DRIVER} ${driver_node} ${res} = Get Subscribe Group ${app_node_id} @@ -147,7 +147,7 @@ Delete the subscribe group config, it should return failure Check Error Code ${res} ${ERR_GRP_CONFIG_IN_USE} North APP unsubscribe non-existent group config, it should return failure - ${app_node_id} = Get Node ID ${NODE_APP} ${app_node} + ${app_node_id} = Get Node ID ${NODE_MQTT} ${app_node} ${driver_node_id} = Get Node ID ${NODE_DRIVER} ${driver_node} ${res} = Unsubscribe Group ${driver_node_id} ${app_node_id} xxgrp_config @@ -155,7 +155,7 @@ North APP unsubscribe non-existent group config, it should return failure Check Error Code ${res} ${ERR_GRPCONFIG_NOT_EXIST} North APP unsubscribe group config, it should return success - ${app_node_id} = Get Node ID ${NODE_APP} ${app_node} + ${app_node_id} = Get Node ID ${NODE_MQTT} ${app_node} ${driver_node_id1} = Get Node ID ${NODE_DRIVER} ${driver_node} ${driver_node_id2} = Get Node ID ${NODE_DRIVER} driver-node2 diff --git a/ft/node_api_test.robot b/ft/node_api_test.robot index c37e3ead..c907fee2 100644 --- a/ft/node_api_test.robot +++ b/ft/node_api_test.robot @@ -71,12 +71,6 @@ Get MQTT node, it should return all mqtt node Node With Name Should Exist ${res}[nodes] mqtt-adapter -Get STREAM PROCESSOR node, it should return all STREAM PROCESSOR node - ${res} = Get Nodes ${NODE_STREAM_PROCESSOR} - - Check Response Status ${res} 200 - Run Keyword If ${res}[nodes] != @{EMPTY} Fail not an empty object - Get UNKNOWN type node, it should return empty node ${res} = Get Nodes ${NODE_UNKNOWN} @@ -163,13 +157,13 @@ Update a node that does not exist, it should return failure Check Error Code ${res} ${ERR_NODE_NOT_EXIST} Update the name of the node, it should return success - ${app_node_id} = Add Node And Return ID ${NODE_APP} app-node ${PLUGIN_MQTT} + ${app_node_id} = Add Node And Return ID ${NODE_MQTT} app-node ${PLUGIN_MQTT} ${res} = Update Node ${app_node_id} test-name Check Response Status ${res} 200 Check Error Code ${res} ${ERR_SUCCESS} - ${res} = Get Nodes ${NODE_APP} + ${res} = Get Nodes ${NODE_MQTT} Check Response Status ${res} 200 Node With Name Should Exist ${res}[nodes] test-name diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 5aaee1ac..7eba1a16 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -87,8 +87,8 @@ typedef struct adapter_callbacks { union { struct { - int (*update)(neu_adapter_t *adapter, const char *tag_name, - neu_value_u value); + int (*update)(neu_adapter_t *adapter, const char *address, + neu_dvalue_t value); } driver; }; } adapter_callbacks_t; diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index dceeb4b5..aa525426 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -24,6 +24,8 @@ extern "C" { #endif +#include "utils/utarray.h" + #include "adapter.h" #include "tag_group_config.h" #include "type.h" @@ -42,8 +44,10 @@ typedef struct neu_plugin_common { typedef struct neu_plugin neu_plugin_t; typedef struct neu_plugin_group { - neu_taggrp_config_t *group; - void * ctx; + char * group_name; + UT_array *tags; + + void *ctx; } neu_plugin_group_t; typedef struct neu_plugin_intf_funs { @@ -67,12 +71,9 @@ typedef struct neu_plugin_intf_funs { neu_plugin_group_t *new_group); void (*del_group)(neu_plugin_t * plugin, neu_plugin_group_t *old_group); - void (*group_group)(neu_plugin_t * plugin, - neu_plugin_group_t *old_group, - neu_plugin_group_t *new_group); int (*group_timer)(neu_plugin_t *plugin, neu_plugin_group_t *group); - int (*write_tag)(neu_plugin_t *plugin, neu_datatag_t *tag, - neu_value_u value); + int (*write_tag)(neu_plugin_t *plugin, uint32_t req_id, + neu_datatag_t *tag, neu_value_u value); } driver; }; diff --git a/include/neuron/type.h b/include/neuron/type.h index 314617ce..3b878f43 100644 --- a/include/neuron/type.h +++ b/include/neuron/type.h @@ -24,6 +24,8 @@ extern "C" { #endif +#include +#include #include typedef enum { @@ -40,8 +42,11 @@ typedef enum { NEU_TYPE_BIT = 11, NEU_TYPE_BOOL = 12, NEU_TYPE_STRING = 13, + NEU_TYPE_BYTES = 14, + NEU_TYPE_ERROR = 15, } neu_type_e; +#define NEU_VALUE_SIZE 128 typedef union { int8_t i8; uint8_t u8; @@ -53,9 +58,337 @@ typedef union { uint64_t u64; float f32; double d64; - char * str; + char str[NEU_VALUE_SIZE]; + uint8_t bytes[NEU_VALUE_SIZE]; } neu_value_u; +typedef struct { + neu_type_e type; + neu_value_u value; +} neu_dvalue_t; +typedef union neu_value8 { + uint8_t value; + struct { + uint8_t b0 : 1; + uint8_t b1 : 1; + uint8_t b2 : 1; + uint8_t b3 : 1; + uint8_t b4 : 1; + uint8_t b5 : 1; + uint8_t b6 : 1; + uint8_t b7 : 1; + } __attribute__((packed)) vb; +} neu_value8_u; + +typedef union neu_value16 { + uint16_t value; + struct { + uint8_t b0 : 1; + uint8_t b1 : 1; + uint8_t b2 : 1; + uint8_t b3 : 1; + uint8_t b4 : 1; + uint8_t b5 : 1; + uint8_t b6 : 1; + uint8_t b7 : 1; + uint8_t b8 : 1; + uint8_t b9 : 1; + uint8_t b10 : 1; + uint8_t b11 : 1; + uint8_t b12 : 1; + uint8_t b13 : 1; + uint8_t b14 : 1; + uint8_t b15 : 1; + } __attribute__((packed)) vb; +} neu_value16_u; + +typedef union neu_value24 { + uint8_t value[3]; +} neu_value24_u; + +typedef union neu_value32 { + uint32_t vint; + float vfloat; +} neu_value32_u; + +typedef union neu_value64 { + uint64_t vint; + double vdouble; +} neu_value64_u; + +static inline void neu_value24_set(union neu_value24 *v24, uint32_t v) +{ + assert((v & 0xff000000) == 0); + + v24->value[0] = v & 0xff; + v24->value[1] = (v >> 8) & 0xff; + v24->value[2] = (v >> 16) & 0xff; +} + +static inline uint32_t neu_value24_get(union neu_value24 v24) +{ + uint32_t ret = 0; + + ret |= v24.value[0]; + ret |= v24.value[1] << 8; + ret |= v24.value[2] << 16; + + return ret; +} + +static inline uint8_t neu_value8_get_bit(neu_value8_u value, uint8_t index) +{ + uint8_t ret = 0; + + switch (index) { + case 0: + ret = value.vb.b0; + break; + case 1: + ret = value.vb.b1; + break; + case 2: + ret = value.vb.b2; + break; + case 3: + ret = value.vb.b3; + break; + case 4: + ret = value.vb.b4; + break; + case 5: + ret = value.vb.b5; + break; + case 6: + ret = value.vb.b6; + break; + case 7: + ret = value.vb.b7; + break; + default: + assert(1 == 0); + break; + } + + return ret; +} + +static inline void neu_value8_set_bit(neu_value8_u *value, uint8_t index, + uint8_t v) +{ + switch (index) { + case 0: + value->vb.b0 = v; + break; + case 1: + value->vb.b1 = v; + break; + case 2: + value->vb.b2 = v; + break; + case 3: + value->vb.b3 = v; + break; + case 4: + value->vb.b4 = v; + break; + case 5: + value->vb.b5 = v; + break; + case 6: + value->vb.b6 = v; + break; + case 7: + value->vb.b7 = v; + break; + default: + assert(1 == 0); + break; + } +} + +static inline uint8_t neu_value16_get_bit(neu_value16_u value, uint8_t index) +{ + uint8_t ret = 0; + + switch (index) { + case 0: + ret = value.vb.b0; + break; + case 1: + ret = value.vb.b1; + break; + case 2: + ret = value.vb.b2; + break; + case 3: + ret = value.vb.b3; + break; + case 4: + ret = value.vb.b4; + break; + case 5: + ret = value.vb.b5; + break; + case 6: + ret = value.vb.b6; + break; + case 7: + ret = value.vb.b7; + break; + case 8: + ret = value.vb.b8; + break; + case 9: + ret = value.vb.b9; + break; + case 10: + ret = value.vb.b10; + break; + case 11: + ret = value.vb.b11; + break; + case 12: + ret = value.vb.b12; + break; + case 13: + ret = value.vb.b13; + break; + case 14: + ret = value.vb.b14; + break; + case 15: + ret = value.vb.b15; + break; + default: + assert(1 == 0); + break; + } + + return ret; +} + +static inline void neu_value16_set_bit(neu_value16_u *value, uint8_t index, + uint8_t v) +{ + switch (index) { + case 0: + value->vb.b0 = v; + break; + case 1: + value->vb.b1 = v; + break; + case 2: + value->vb.b2 = v; + break; + case 3: + value->vb.b3 = v; + break; + case 4: + value->vb.b4 = v; + break; + case 5: + value->vb.b5 = v; + break; + case 6: + value->vb.b6 = v; + break; + case 7: + value->vb.b7 = v; + break; + case 8: + value->vb.b8 = v; + break; + case 9: + value->vb.b9 = v; + break; + case 10: + value->vb.b10 = v; + break; + case 11: + value->vb.b11 = v; + break; + case 12: + value->vb.b12 = v; + break; + case 13: + value->vb.b13 = v; + break; + case 14: + value->vb.b14 = v; + break; + case 15: + value->vb.b15 = v; + break; + default: + assert(1 == 0); + break; + } +} + +static inline void neu_ntohs24_p(uint8_t *value) +{ + uint8_t t = 0; + + t = value[0]; + value[0] = value[2]; + value[2] = t; +} + +static inline void neu_htons24_p(uint8_t *value) +{ + neu_ntohs24_p(value); +} + +static inline uint64_t neu_htonll(uint64_t u) +{ + uint8_t *bytes = (uint8_t *) &u; + uint64_t ret = 0; + uint8_t *ret_bytes = (uint8_t *) &ret; + + ret_bytes[0] = bytes[3]; + ret_bytes[1] = bytes[2]; + ret_bytes[2] = bytes[1]; + ret_bytes[3] = bytes[0]; + + return ret; +} + +static inline uint64_t neu_ntohll(uint64_t u) +{ + return neu_htonll(u); +} + +static inline void neu_ntonll_p(uint64_t *pu) +{ + *pu = neu_htonll(*pu); +} + +static inline void neu_htonll_p(uint64_t *pu) +{ + neu_ntonll_p(pu); +} + +static inline void neu_ntohl_p(uint32_t *pu) +{ + *pu = ntohl(*pu); +} + +static inline void neu_htonl_p(uint32_t *pu) +{ + neu_ntohl_p(pu); +} + +static inline void neu_ntohs_p(uint16_t *pu) +{ + *pu = ntohs(*pu); +} + +static inline void neu_htons_p(uint16_t *pu) +{ + neu_ntohs_p(pu); +} + #ifdef __cplusplus } #endif diff --git a/plugins/restful/license.c b/plugins/restful/license.c index ad8b390e..633968c0 100644 --- a/plugins/restful/license.c +++ b/plugins/restful/license.c @@ -166,6 +166,7 @@ static const char *const g_plugin_bit_str_map[] = { [PLUGIN_BIT_S7COMM] = "s7comm", [PLUGIN_BIT_FINS] = "fins", [PLUGIN_BIT_QNA3E] = "QnA3E", + [PLUGIN_BIT_IEC60870_5_104] = "IEC60870-5-104", [PLUGIN_BIT_MAX] = NULL, }; diff --git a/plugins/restful/license.h b/plugins/restful/license.h index a5280966..c86ec756 100644 --- a/plugins/restful/license.h +++ b/plugins/restful/license.h @@ -77,6 +77,7 @@ typedef enum { PLUGIN_BIT_S7COMM = 3, PLUGIN_BIT_FINS = 4, PLUGIN_BIT_QNA3E = 5, + PLUGIN_BIT_IEC60870_5_104 = 6, PLUGIN_BIT_MAX, } plugin_bit_e; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 38b3f47b..bfd79c89 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -979,7 +979,7 @@ static void adapter_loop(void *arg) req.buf = (void *) &sub_node_req; if (adapter->type == ADAPTER_TYPE_DRIVERX) { neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - pay_msg_type, &req); + &req); } else { intf_funs->request(adapter->plugin, &req); } @@ -1006,7 +1006,7 @@ static void adapter_loop(void *arg) req.buf = (void *) &unsub_node_req; if (adapter->type == ADAPTER_TYPE_DRIVERX) { neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - pay_msg_type, &req); + &req); } else { intf_funs->request(adapter->plugin, &req); } @@ -1032,7 +1032,7 @@ static void adapter_loop(void *arg) req.buf = (void *) &read_req; if (adapter->type == ADAPTER_TYPE_DRIVERX) { neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - pay_msg_type, &req); + &req); } else { intf_funs->request(adapter->plugin, &req); } @@ -1102,7 +1102,7 @@ static void adapter_loop(void *arg) req.buf = (void *) &write_req; if (adapter->type == ADAPTER_TYPE_DRIVERX) { neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - pay_msg_type, &req); + &req); } else { intf_funs->request(adapter->plugin, &req); } @@ -1839,6 +1839,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, adapter->plugin_kind = info->plugin_kind; adapter->manager = manager; adapter->trans_kind = NEURON_TRANS_DATAVAL; + adapter->node_id = neu_manager_adapter_id_to_node_id(manager, info->id); if (adapter->name == NULL || info->plugin_lib_name == NULL) { if (adapter->name != NULL) { @@ -1861,10 +1862,15 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, info->plugin_lib_name, adapter->name); } + adapter->cb_funs = callback_funs; + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + adapter = (neu_adapter_t *) neu_adapter_driver_create(adapter); + } + neu_plugin_t *plugin; adapter->plugin_lib = handle; adapter->plugin_module = plugin_module; - plugin = plugin_module->intf_funs->open(adapter, &callback_funs); + plugin = plugin_module->intf_funs->open(adapter, &adapter->cb_funs); if (plugin == NULL) { log_error("Can't to open plugin(%s)", plugin_module->module_name); free(adapter->name); @@ -1890,10 +1896,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, nng_setopt(adapter->sock, NNG_OPT_RECVTIMEO, &recv_timeout, sizeof(recv_timeout)); - if (adapter->type == ADAPTER_TYPE_DRIVERX) { - adapter = (neu_adapter_t *) neu_adapter_driver_create(adapter); - } - log_info("Success to create adapter: %s", adapter->name); return adapter; } @@ -1940,6 +1942,10 @@ int neu_adapter_init(neu_adapter_t *adapter) return (-1); } + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_init((neu_adapter_driver_t *) adapter); + } + if (adapter->plugin_module != NULL) { const neu_plugin_intf_funs_t *intf_funs; intf_funs = adapter->plugin_module->intf_funs; @@ -1961,6 +1967,10 @@ int neu_adapter_uninit(neu_adapter_t *adapter) return -1; } + if (adapter->type == ADAPTER_TYPE_DRIVERX) { + neu_adapter_driver_init((neu_adapter_driver_t *) adapter); + } + log_info("Stop the adapter(%s)", adapter->name); nng_mtx_lock(adapter->mtx); adapter->stop = true; diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index fc45e46f..dc46a9a0 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -43,6 +43,7 @@ typedef enum adapter_state { struct neu_adapter { adapter_id_t id; + neu_node_id_t node_id; adapter_type_e type; nng_mtx * mtx; nng_mtx * sub_grp_mtx; diff --git a/src/adapter/cache.c b/src/adapter/cache.c index 2b545f25..048232e0 100644 --- a/src/adapter/cache.c +++ b/src/adapter/cache.c @@ -62,6 +62,9 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache) HASH_ITER(hh, cache->table, elem, tmp) { HASH_DEL(cache->table, elem); + if (elem->n_byte > 0) { + free(elem->bytes); + } free(elem); } pthread_mutex_unlock(&cache->mtx); @@ -79,10 +82,13 @@ void neu_driver_cache_error(neu_driver_cache_t *cache, const char *key, pthread_mutex_lock(&cache->mtx); HASH_FIND_STR(cache->table, key, elem); - if (elem != NULL) { - elem->timestamp = timestamp; - elem->error = error; + if (elem == NULL) { + elem = calloc(1, sizeof(struct elem)); + strncpy(elem->name, key, sizeof(elem->name)); + HASH_ADD_STR(cache->table, name, elem); } + elem->timestamp = timestamp; + elem->error = error; pthread_mutex_unlock(&cache->mtx); } @@ -99,7 +105,7 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key, elem = calloc(1, sizeof(struct elem)); strncpy(elem->name, key, sizeof(elem->name)); assert(n_byte != 0); - assert(n_byte <= NEU_DRIVER_VALUE_SIZE); + assert(n_byte <= NEU_VALUE_SIZE); elem->n_byte = n_byte; elem->bytes = calloc(elem->n_byte, 1); HASH_ADD_STR(cache->table, name, elem); @@ -108,7 +114,15 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key, elem->error = 0; elem->timestamp = timestamp; - assert(n_byte == elem->n_byte); + if (elem->n_byte == 0) { + assert(n_byte != 0); + assert(n_byte <= NEU_VALUE_SIZE); + elem->n_byte = n_byte; + elem->bytes = calloc(elem->n_byte, 1); + } else { + assert(n_byte == elem->n_byte); + } + memcpy(elem->bytes, bytes, elem->n_byte); pthread_mutex_unlock(&cache->mtx); @@ -127,7 +141,7 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *key, value->error = elem->error; value->n_byte = elem->n_byte; value->timestamp = elem->timestamp; - memcpy(value->bytes, elem->bytes, elem->n_byte); + memcpy(value->value.bytes, elem->bytes, elem->n_byte); ret = 0; } diff --git a/src/adapter/cache.h b/src/adapter/cache.h index d9736d41..734caafa 100644 --- a/src/adapter/cache.h +++ b/src/adapter/cache.h @@ -22,6 +22,8 @@ #include +#include "type.h" + typedef struct neu_driver_cache neu_driver_cache_t; neu_driver_cache_t *neu_driver_cache_new(); @@ -34,10 +36,9 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *key, uint8_t *bytes); void neu_driver_cache_del(neu_driver_cache_t *cache, const char *key); -#define NEU_DRIVER_VALUE_SIZE 256 typedef struct { - uint16_t n_byte; - uint8_t bytes[NEU_DRIVER_VALUE_SIZE]; + uint16_t n_byte; + neu_value_u value; int32_t error; int64_t timestamp; diff --git a/src/adapter/driver.c b/src/adapter/driver.c index c518b126..e85f2aa6 100644 --- a/src/adapter/driver.c +++ b/src/adapter/driver.c @@ -17,29 +17,98 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ #include +#include +#include #include #include "event/event.h" -#include "utils/utarray.h" +#include "utils/uthash.h" #include "adapter.h" #include "adapter_internal.h" #include "cache.h" #include "driver_internal.h" +struct subscribe_group { + neu_plugin_group_t grp; + neu_taggrp_config_t *group; + + char *name; + + neu_event_timer_t *report; + neu_event_timer_t *read; + + neu_adapter_driver_t *driver; + + UT_hash_handle hh; +}; + struct neu_adapter_driver { - neu_adapter_t adapter; + neu_adapter_t adapter; + neu_datatag_table_t *tag_table; neu_driver_cache_t *cache; neu_events_t *msg_events; neu_events_t *driver_events; -}; -// read/write/subscribe/unsubscribe -struct subscribe_group { + pthread_mutex_t grp_mtx; + struct subscribe_group *grps; }; +static int report_callback(void *usr_data); +static int read_callback(void *usr_data); +static neu_data_val_t *read_group(neu_adapter_driver_t *driver, + neu_taggrp_config_t * grp); +static void load_tags(neu_adapter_driver_t *driver, neu_taggrp_config_t *group, + UT_array **tags); +static void free_tags(UT_array *tags); + +static int update(neu_adapter_t *adapter, const char *address, + neu_dvalue_t value) +{ + neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter; + uint16_t n_byte = 0; + + switch (value.type) { + case NEU_TYPE_INT8: + case NEU_TYPE_UINT8: + case NEU_TYPE_BIT: + case NEU_TYPE_BOOL: + n_byte = 1; + break; + case NEU_TYPE_INT16: + case NEU_TYPE_UINT16: + n_byte = 2; + break; + case NEU_TYPE_INT32: + case NEU_TYPE_UINT32: + case NEU_TYPE_FLOAT: + n_byte = 4; + break; + case NEU_TYPE_INT64: + case NEU_TYPE_UINT64: + case NEU_TYPE_DOUBLE: + n_byte = 8; + break; + case NEU_TYPE_STRING: + n_byte = strlen(value.value.str); + break; + case NEU_TYPE_BYTES: + break; + default: + break; + } + + if (value.type == NEU_TYPE_ERROR) { + neu_driver_cache_error(driver->cache, address, 0, value.value.i32); + } else { + neu_driver_cache_update(driver->cache, address, 0, n_byte, + value.value.bytes); + } + return 0; +} + neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter) { neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t)); @@ -47,9 +116,11 @@ neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter) driver->adapter = *adapter; free(adapter); - driver->cache = neu_driver_cache_new(); - driver->msg_events = neu_event_new(); - driver->driver_events = neu_event_new(); + pthread_mutex_init(&driver->grp_mtx, NULL); + driver->cache = neu_driver_cache_new(); + driver->msg_events = neu_event_new(); + driver->driver_events = neu_event_new(); + driver->adapter.cb_funs.driver.update = update; return driver; } @@ -75,43 +146,142 @@ int neu_adapter_driver_stop(neu_adapter_driver_t *driver) int neu_adapter_driver_init(neu_adapter_driver_t *driver) { - (void) driver; + driver->tag_table = neu_manager_get_datatag_tbl(driver->adapter.manager, + driver->adapter.node_id); + return 0; } int neu_adapter_driver_uninit(neu_adapter_driver_t *driver) { - (void) driver; + struct subscribe_group *sg = NULL; + struct subscribe_group *tmp = NULL; + + pthread_mutex_lock(&driver->grp_mtx); + HASH_ITER(hh, driver->grps, sg, tmp) + { + driver->adapter.plugin_module->intf_funs->driver.del_group( + driver->adapter.plugin, &sg->grp); + HASH_DEL(driver->grps, sg); + neu_event_del_timer(driver->msg_events, sg->report); + neu_event_del_timer(driver->driver_events, sg->read); + free(sg); + } + + pthread_mutex_unlock(&driver->grp_mtx); + return 0; } void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver, - msg_type_e type, neu_request_t *req) + neu_request_t * req) { - (void) driver; - switch (type) { - case MSG_CMD_READ_DATA: { - neu_reqresp_read_t *read_req = (neu_reqresp_read_t *) req->buf; + switch (req->req_type) { + case NEU_REQRESP_READ_DATA: { + neu_reqresp_read_t * read_req = (neu_reqresp_read_t *) req->buf; + neu_reqresp_read_resp_t data = { + .grp_config = read_req->grp_config, + .data_val = read_group(driver, read_req->grp_config), + }; + neu_response_t resp = { + .req_id = req->req_id, + .resp_type = NEU_REQRESP_READ_RESP, + .buf = &data, + .buf_len = sizeof(neu_reqresp_read_resp_t), + .recver_id = req->sender_id, + }; - (void) read_req->grp_config; + driver->adapter.cb_funs.response(&driver->adapter, &resp); + break; + } + case NEU_REQRESP_WRITE_DATA: { + neu_reqresp_write_t *write_req = (neu_reqresp_write_t *) req->buf; + neu_fixed_array_t * array = NULL; + neu_int_val_t * int_val = NULL; + neu_value_u value = { 0 }; + neu_datatag_t * tag = NULL; + + neu_dvalue_get_ref_array(write_req->data_val, &array); + assert(array->length == 1); + + int_val = neu_fixed_array_get(array, 0); + + tag = neu_datatag_tbl_get(driver->tag_table, int_val->key); + if (tag == NULL) { + int error = NEU_ERR_TAG_NOT_EXIST; + neu_reqresp_write_resp_t data = { + .grp_config = write_req->grp_config, + .data_val = neu_datatag_pack_create(1), + }; + neu_response_t resp = { + .req_id = req->req_id, + .resp_type = NEU_REQRESP_WRITE_RESP, + .buf = &data, + .buf_len = sizeof(neu_reqresp_write_resp_t), + .recver_id = req->sender_id, + }; + + neu_datatag_pack_add(data.data_val, 0, NEU_DTYPE_ERRORCODE, + int_val->key, (void *) &error); + driver->adapter.cb_funs.response(&driver->adapter, &resp); + } else { + driver->adapter.plugin_module->intf_funs->driver.write_tag( + driver->adapter.plugin, req->req_id, tag, value); + } break; } - case MSG_CMD_WRITE_DATA: - break; - case MSG_CMD_SUBSCRIBE_NODE: { + case NEU_REQRESP_SUBSCRIBE_NODE: { neu_reqresp_subscribe_node_t *sub_req = (neu_reqresp_subscribe_node_t *) req->buf; + uint32_t interval = neu_taggrp_cfg_get_interval(sub_req->grp_config); + struct subscribe_group *sg = calloc(1, sizeof(struct subscribe_group)); + neu_event_timer_param_t param = { + .second = interval / 1000, + .millisecond = interval % 1000, + .usr_data = sg, + }; - (void) sub_req->grp_config; + sg->driver = driver; + sg->group = sub_req->grp_config; + // strncpy(sg->name, neu_taggrp_cfg_get_name(sg->group), + // sizeof(sg->name)); + sg->name = strdup(neu_taggrp_cfg_get_name(sg->group)); + sg->grp.group_name = strdup(neu_taggrp_cfg_get_name(sg->group)); + load_tags(driver, sg->group, &sg->grp.tags); + driver->adapter.plugin_module->intf_funs->driver.add_group( + driver->adapter.plugin, &sg->grp); + + pthread_mutex_lock(&driver->grp_mtx); + HASH_ADD_STR(driver->grps, name, sg); + + param.cb = report_callback; + sg->report = neu_event_add_timer(driver->msg_events, param); + param.cb = read_callback; + sg->read = neu_event_add_timer(driver->driver_events, param); + pthread_mutex_unlock(&driver->grp_mtx); break; } - case MSG_CMD_UNSUBSCRIBE_NODE: { + case NEU_REQRESP_UNSUBSCRIBE_NODE: { neu_reqresp_unsubscribe_node_t *unsub_req = (neu_reqresp_unsubscribe_node_t *) req->buf; + struct subscribe_group *sg = NULL; - (void) unsub_req->grp_config; + pthread_mutex_lock(&driver->grp_mtx); + HASH_FIND_STR(driver->grps, + neu_taggrp_cfg_get_name(unsub_req->grp_config), sg); + if (sg != NULL) { + driver->adapter.plugin_module->intf_funs->driver.del_group( + driver->adapter.plugin, &sg->grp); + HASH_DEL(driver->grps, sg); + neu_event_del_timer(driver->msg_events, sg->report); + neu_event_del_timer(driver->driver_events, sg->read); + free(sg); + } + // assert(sg != NULL); + + pthread_mutex_unlock(&driver->grp_mtx); break; } @@ -120,3 +290,183 @@ void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver, break; } } + +static int report_callback(void *usr_data) +{ + struct subscribe_group *sg = (struct subscribe_group *) usr_data; + + pthread_mutex_lock(&sg->driver->grp_mtx); + neu_reqresp_data_t data = { + .grp_config = sg->group, + .data_val = read_group(sg->driver, sg->group), + }; + neu_response_t resp = { + .req_id = 1, + .resp_type = NEU_REQRESP_TRANS_DATA, + .buf = &data, + .buf_len = sizeof(neu_reqresp_data_t), + }; + + sg->driver->adapter.cb_funs.response(&sg->driver->adapter, &resp); + pthread_mutex_unlock(&sg->driver->grp_mtx); + + return 0; +} + +static int read_callback(void *usr_data) +{ + struct subscribe_group *sg = (struct subscribe_group *) usr_data; + + pthread_mutex_lock(&sg->driver->grp_mtx); + if (neu_taggrp_cfg_is_anchored(sg->group)) { + sg->driver->adapter.plugin_module->intf_funs->driver.group_timer( + sg->driver->adapter.plugin, &sg->grp); + } else { + neu_taggrp_config_t *new_config = neu_system_find_group_config( + sg->driver->adapter.plugin, sg->driver->adapter.node_id, + neu_taggrp_cfg_get_name(sg->group)); + assert(new_config != NULL); + + sg->driver->adapter.plugin_module->intf_funs->driver.del_group( + sg->driver->adapter.plugin, &sg->grp); + + neu_plugin_group_t grp = { .group_name = sg->grp.group_name }; + load_tags(sg->driver, new_config, &grp.tags); + + sg->driver->adapter.plugin_module->intf_funs->driver.add_group( + sg->driver->adapter.plugin, &grp); + + sg->group = new_config; + free_tags(sg->grp.tags); + sg->grp = grp; + } + pthread_mutex_unlock(&sg->driver->grp_mtx); + + return 0; +} + +static neu_data_val_t *read_group(neu_adapter_driver_t *driver, + neu_taggrp_config_t * grp) +{ + vector_t * ids = neu_taggrp_cfg_get_datatag_ids(grp); + neu_data_val_t *val = neu_datatag_pack_create(ids->size); + int index = -1; + + VECTOR_FOR_EACH(ids, iter) + { + + int error = NEU_ERR_SUCCESS; + neu_driver_cache_value_t value = { 0 }; + neu_datatag_id_t * id = (neu_datatag_id_t *) iterator_get(&iter); + neu_datatag_t *tag = neu_datatag_tbl_get(driver->tag_table, *id); + + index += 1; + if (tag == NULL) { + error = NEU_ERR_TAG_NOT_EXIST; + neu_datatag_pack_add(val, index, NEU_DTYPE_ERRORCODE, *id, + (void *) &error); + continue; + } + + if ((tag->attribute & NEU_ATTRIBUTE_READ) != NEU_ATTRIBUTE_READ) { + error = NEU_ERR_PLUGIN_TAG_NOT_ALLOW_READ; + neu_datatag_pack_add(val, index, NEU_DTYPE_ERRORCODE, tag->id, + (void *) &error); + continue; + } + + if (neu_driver_cache_get(driver->cache, tag->addr_str, &value) != 0) { + error = NEU_ERR_TAG_NOT_EXIST; + neu_datatag_pack_add(val, index, NEU_DTYPE_ERRORCODE, *id, + (void *) &error); + continue; + } + + if (value.error != 0) { + neu_datatag_pack_add(val, index, NEU_DTYPE_ERRORCODE, *id, + (void *) &value.error); + continue; + } + + switch (tag->type) { + case NEU_DTYPE_BOOL: + case NEU_DTYPE_BIT: + case NEU_DTYPE_INT8: + case NEU_DTYPE_UINT8: + assert(value.n_byte == sizeof(uint8_t)); + + neu_datatag_pack_add(val, index, tag->type, tag->id, + (void *) &value.value.u8); + break; + case NEU_DTYPE_INT16: + case NEU_DTYPE_UINT16: { + assert(value.n_byte == sizeof(uint16_t)); + uint16_t v = value.value.u16; + + v = ntohs(v); + + neu_datatag_pack_add(val, index, tag->type, tag->id, (void *) &v); + break; + } + case NEU_DTYPE_FLOAT: + case NEU_DTYPE_INT32: + case NEU_DTYPE_UINT32: { + assert(value.n_byte == sizeof(uint32_t)); + uint32_t v = value.value.u32; + + v = ntohl(v); + + neu_datatag_pack_add(val, index, tag->type, tag->id, (void *) &v); + break; + } + case NEU_DTYPE_INT64: + case NEU_DTYPE_UINT64: + case NEU_DTYPE_DOUBLE: { + assert(value.n_byte == sizeof(uint64_t)); + uint64_t v = value.value.u64; + + v = neu_ntohll(v); + + neu_datatag_pack_add(val, index, tag->type, tag->id, (void *) &v); + break; + } + case NEU_DTYPE_CSTR: + if (neu_datatag_string_is_utf8(value.value.str, value.n_byte)) { + neu_datatag_pack_add(val, index, tag->type, tag->id, + (void *) value.value.str); + } else { + char *unknown = "?"; + neu_datatag_pack_add(val, index, tag->type, tag->id, + (void *) unknown); + } + break; + default: + assert(tag->type == 0); + break; + } + } + + return val; +} + +static void load_tags(neu_adapter_driver_t *driver, neu_taggrp_config_t *group, + UT_array **tags) +{ + vector_t *ids = neu_taggrp_cfg_get_datatag_ids(group); + + utarray_new(*tags, &ut_ptr_icd); + + VECTOR_FOR_EACH(ids, iter) + { + + neu_datatag_id_t *id = (neu_datatag_id_t *) iterator_get(&iter); + neu_datatag_t * tag = neu_datatag_tbl_get(driver->tag_table, *id); + + utarray_push_back(*tags, (void *) &tag); + } +} + +static void free_tags(UT_array *tags) +{ + utarray_free(tags); +} \ No newline at end of file diff --git a/src/adapter/driver_internal.h b/src/adapter/driver_internal.h index 206c3a16..9a9087de 100644 --- a/src/adapter/driver_internal.h +++ b/src/adapter/driver_internal.h @@ -32,6 +32,6 @@ int neu_adapter_driver_init(neu_adapter_driver_t *driver); int neu_adapter_driver_uninit(neu_adapter_driver_t *driver); void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver, - msg_type_e type, neu_request_t *req); + neu_request_t * req); #endif diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index 25ad927b..7bb2b49c 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -389,8 +389,8 @@ static neu_err_code_e manager_reg_adapter(neu_manager_t * manager, } adapter_info.id = manager_new_adapter_id(manager); - adapter_info.type = reg_param->adapter_type; adapter_info.name = reg_param->adapter_name; + adapter_info.type = plugin_reg_info.adapter_type; adapter_info.plugin_id = plugin_reg_info.plugin_id; adapter_info.plugin_kind = plugin_reg_info.plugin_kind; adapter_info.plugin_lib_name = plugin_reg_info.plugin_lib_name;