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

Merge pull request #421 from fengzeroz/main

Browse Source 
Add driver type adapter, simplify driver development
This commit is contained in:
fengzeroz
2022-04-13 17:53:02 +08:00
committed by GitHub
parent 5c557406a0 54a5a2b9d4
commit ccfcc319de
15 changed files with 772 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/common.resource
+1 -1
View File
@@ -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
ft/grp_config_api_test.robot
+5 -5
View File
@@ -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
ft/node_api_test.robot
+2 -8
View File
@@ -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
include/neuron/adapter.h
+2 -2
View File
@@ -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;
include/neuron/plugin.h
+8 -7
View File
@@ -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;
};
include/neuron/type.h
+334 -1
View File
@@ -24,6 +24,8 @@
extern "C" {
#endif
#include <assert.h>
#include <netinet/in.h>
#include <stdint.h>
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
plugins/restful/license.c
+1
View File
@@ -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,
};
plugins/restful/license.h
+1
View File
@@ -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;
src/adapter/adapter.c
+19 -9
View File
@@ -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;
src/adapter/adapter_internal.h
+1
View File
@@ -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;
src/adapter/cache.c
+20 -6
View File
@@ -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;
}
src/adapter/cache.h
+4 -3
View File
@@ -22,6 +22,8 @@
#include <stdint.h>
#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;
src/adapter/driver.c
+372 -22
View File
@@ -17,29 +17,98 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <assert.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdlib.h>
#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);
}
src/adapter/driver_internal.h
+1 -1
View File
@@ -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
src/core/neu_manager.c
+1 -1
View File
@@ -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;