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

Add cyclic reading support and subscription support

Browse Source
This commit is contained in:
gc87
2021-09-24 16:10:03 +08:00
committed by Parker Liu
parent eb00cc6070
commit 5898c2ab8e
5 changed files with 215 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/opcua/open62541_client.c
+1 -1
View File
@@ -361,7 +361,7 @@ int open62541_client_read(open62541_client_t *client, vector_t *data)
UA_ReadResponse resp = UA_Client_Service_read(client->ua_client, req);
uint32_t status_code = resp.responseHeader.serviceResult;
if (UA_STATUSCODE_GOOD != status_code || count != (int) resp.resultsSize) {
log_error("Respons status code:%d", status_code);
log_error("Respons status code:%x", status_code);
return -4;
}
plugins/opcua/plugin_handle.c
+145 -1
View File
@@ -26,6 +26,9 @@
#define UNUSED(x) (void) (x)
static void start_periodic_read(opc_subscribe_tuple_t *tuple);
static void stop_periodic_read(opc_subscribe_tuple_t *tuple);
static neu_datatag_t *get_datatag_by_id(neu_datatag_table_t *table,
datatag_id_t id)
{
@@ -232,6 +235,7 @@ int plugin_handle_read_once(opc_handle_context_t *context,
int rc = open62541_client_is_connected(context->client);
if (0 != rc) {
log_error("open62541 client offline");
vector_uninit(&data);
return -1;
}
@@ -249,9 +253,149 @@ int plugin_handle_write_value(opc_handle_context_t *context,
int rc = open62541_client_is_connected(context->client);
if (0 != rc) {
log_error("open62541 client offline");
return -1;
}
rc = open62541_client_write(context->client, data);
return 0;
}
}
static void periodic_read(nng_aio *aio, void *arg, int code)
{
switch (code) {
case NNG_ETIMEDOUT: {
opc_subscribe_tuple_t *tuple = (opc_subscribe_tuple_t *) arg;
if (neu_taggrp_cfg_is_anchored(tuple->config)) {
neu_variable_t *head = neu_variable_create();
plugin_handle_read_once(tuple->context, tuple->config, head);
if (NULL != tuple->periodic_cb) {
tuple->periodic_cb(tuple->plugin, tuple->config, head);
}
neu_variable_destroy(head);
nng_aio_defer(aio, periodic_read, arg);
} else {
neu_taggrp_config_t *new_config = neu_system_find_group_config(
tuple->plugin, tuple->node_id, tuple->name);
stop_periodic_read(tuple);
if (new_config != NULL) {
neu_taggrp_cfg_free(tuple->config);
tuple->config = new_config;
start_periodic_read(tuple);
}
}
break;
}
case NNG_ECANCELED:
log_warn("aio: %p cancel", aio);
break;
default:
log_warn("aio: %p, skip error: %d", aio, code);
break;
}
}
static opc_subscribe_tuple_t *add_subscribe(opc_handle_context_t *context,
neu_taggrp_config_t * config)
{
opc_subscribe_tuple_t *opc_subscribe =
malloc(sizeof(opc_subscribe_tuple_t));
opc_subscribe->plugin = context->plugin;
opc_subscribe->node_id = context->self_node_id;
opc_subscribe->name = strdup(neu_taggrp_cfg_get_name(config));
opc_subscribe->config = config;
NEU_LIST_NODE_INIT(&opc_subscribe->node);
neu_list_append(&context->subscribe_list, opc_subscribe);
return opc_subscribe;
}
static void start_periodic_read(opc_subscribe_tuple_t *tuple)
{
uint32_t interval = neu_taggrp_cfg_get_interval(tuple->config);
nng_aio_alloc(&tuple->aio, NULL, NULL);
nng_aio_set_timeout(tuple->aio, interval != 0 ? interval : 10000);
nng_aio_defer(tuple->aio, periodic_read, tuple);
}
static void start_opc_subscribe(opc_handle_context_t *context,
neu_taggrp_config_t * config)
{
UNUSED(config);
open62541_client_subscribe(context->client, NULL);
}
int plugin_handle_subscribe(opc_handle_context_t * context,
neu_taggrp_config_t * config,
periodic_response_callback periodic_cb,
subscribe_response_callback subscribe_cb)
{
opc_subscribe_tuple_t *tuple = add_subscribe(context, config);
if (NULL == tuple) {
return -1;
}
tuple->periodic_cb = periodic_cb;
tuple->subscribe_cb = subscribe_cb;
start_periodic_read(tuple);
start_opc_subscribe(context, config);
return 0;
}
static opc_subscribe_tuple_t *find_subscribe(opc_handle_context_t *context,
neu_taggrp_config_t * config)
{
opc_subscribe_tuple_t *item;
opc_subscribe_tuple_t *tuple = NULL;
NEU_LIST_FOREACH(&context->subscribe_list, item)
{
if (item->config == config) {
tuple = item;
}
}
return tuple;
}
static void stop_periodic_read(opc_subscribe_tuple_t *tuple)
{
nng_aio_cancel(tuple->aio);
nng_aio_free(tuple->aio);
}
static void remove_subscribe(opc_subscribe_tuple_t *tuple)
{
if (NULL == tuple) {
return;
}
if (NULL != tuple->name) {
free(tuple->name);
}
stop_periodic_read(tuple);
neu_list_remove(&tuple->context->subscribe_list, tuple);
}
static void stop_opc_subscribe(opc_handle_context_t *context,
neu_taggrp_config_t * config)
{
UNUSED(config);
open62541_client_unsubscribe(context->client, NULL);
}
int plugin_handle_unsubscribe(opc_handle_context_t *context,
neu_taggrp_config_t * config)
{
opc_subscribe_tuple_t *tuple = find_subscribe(context, config);
stop_periodic_read(tuple);
stop_opc_subscribe(context, config);
remove_subscribe(tuple);
return 0;
}
plugins/opcua/plugin_handle.h
+30
View File
@@ -24,17 +24,47 @@
extern "C" {
#endif
#include <neuron.h>
#include <nng/nng.h>
typedef void (*periodic_response_callback)(neu_plugin_t * plugin,
neu_taggrp_config_t *config,
neu_variable_t * array);
typedef void (*subscribe_response_callback)(neu_plugin_t * plugin,
neu_taggrp_config_t *config,
neu_variable_t * array);
typedef struct {
neu_plugin_t * plugin;
open62541_client_t * client;
neu_datatag_table_t *table;
neu_node_id_t self_node_id;
neu_list subscribe_list;
} opc_handle_context_t;
typedef struct {
neu_list_node node;
neu_plugin_t * plugin;
neu_node_id_t node_id;
neu_taggrp_config_t * config;
char * name;
nng_aio * aio;
periodic_response_callback periodic_cb;
subscribe_response_callback subscribe_cb;
opc_handle_context_t * context;
} opc_subscribe_tuple_t;
int plugin_handle_read_once(opc_handle_context_t *context,
neu_taggrp_config_t *config, neu_variable_t *array);
int plugin_handle_write_value(opc_handle_context_t *context,
neu_variable_t *array, vector_t *data);
int plugin_handle_subscribe(opc_handle_context_t * context,
neu_taggrp_config_t * config,
periodic_response_callback periodic_cb,
subscribe_response_callback subscribe_cb);
int plugin_handle_unsubscribe(opc_handle_context_t *context,
neu_taggrp_config_t * config);
#ifdef __cplusplus
}
plugins/opcua/plugin_opcua.c
+38 -1
View File
@@ -112,8 +112,12 @@ static int opcua_plugin_init(neu_plugin_t *plugin)
memset(plugin->handle_context, 0, sizeof(opc_handle_context_t));
plugin->handle_context->plugin = plugin;
plugin->handle_context->client = plugin->client;
plugin->handle_context->table = NULL;
plugin->handle_context->self_node_id = neu_plugin_self_node_id(plugin);
NEU_LIST_INIT(&plugin->handle_context->subscribe_list,
opc_subscribe_tuple_t, node);
log_info("Initialize plugin: %s", neu_plugin_module.module_name);
return 0;
}
@@ -137,6 +141,27 @@ static int opcua_plugin_config(neu_plugin_t *plugin, neu_config_t *configs)
return rv;
}
static void periodic_response(neu_plugin_t *plugin, neu_taggrp_config_t *config,
neu_variable_t *array)
{
UNUSED(plugin);
UNUSED(config);
UNUSED(array);
// neu_variable_t *head = neu_variable_create();
// neu_plugin_response_trans_data(plugin, config, head, 0);
// neu_variable_destroy(head);
}
static void subscribe_response(neu_plugin_t * plugin,
neu_taggrp_config_t *config,
neu_variable_t * array)
{
UNUSED(plugin);
UNUSED(config);
UNUSED(array);
// TODO: return subscription data
}
static int opcua_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
{
if (plugin == NULL || req == NULL) {
@@ -161,7 +186,6 @@ static int opcua_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
head);
neu_plugin_response_trans_data(plugin, read_req->grp_config, head,
req->req_id);
neu_variable_destroy(head);
break;
}
case NEU_REQRESP_WRITE_DATA: {
@@ -173,6 +197,19 @@ static int opcua_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
vector_uninit(&v);
break;
}
case NEU_REQRESP_SUBSCRIBE_NODE: {
neu_reqresp_read_t *subscribe_req = (neu_reqresp_read_t *) req->buf;
plugin_handle_subscribe(plugin->handle_context,
subscribe_req->grp_config, periodic_response,
subscribe_response);
break;
}
case NEU_REQRESP_UNSUBSCRIBE_NODE: {
neu_reqresp_read_t *unsubscribe_req = (neu_reqresp_read_t *) req->buf;
plugin_handle_unsubscribe(plugin->handle_context,
unsubscribe_req->grp_config);
break;
}
default:
break;
plugins/sample/plugin_sample_drv.c
+1 -1
View File
@@ -149,7 +149,7 @@ static int sample_drv_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
resp.buf_len = sizeof(neu_reqresp_data_t);
resp.buf = &data_resp;
rv = adapter_callbacks->response(plugin->common.adapter, &resp);
neu_variable_destroy(head);
// neu_variable_destroy(head);
break;
}