From 5898c2ab8eecd8bcb778ae6d248fae2413fce26a Mon Sep 17 00:00:00 2001 From: gc87 Date: Wed, 22 Sep 2021 18:12:17 +0800 Subject: [PATCH] Add cyclic reading support and subscription support --- plugins/opcua/open62541_client.c | 2 +- plugins/opcua/plugin_handle.c | 146 ++++++++++++++++++++++++++++- plugins/opcua/plugin_handle.h | 30 ++++++ plugins/opcua/plugin_opcua.c | 39 +++++++- plugins/sample/plugin_sample_drv.c | 2 +- 5 files changed, 215 insertions(+), 4 deletions(-) diff --git a/plugins/opcua/open62541_client.c b/plugins/opcua/open62541_client.c index fee8d62b..711ec9a7 100644 --- a/plugins/opcua/open62541_client.c +++ b/plugins/opcua/open62541_client.c @@ -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; } diff --git a/plugins/opcua/plugin_handle.c b/plugins/opcua/plugin_handle.c index c2532b45..a8b6ea50 100644 --- a/plugins/opcua/plugin_handle.c +++ b/plugins/opcua/plugin_handle.c @@ -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; -} \ No newline at end of file +} + +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; +} diff --git a/plugins/opcua/plugin_handle.h b/plugins/opcua/plugin_handle.h index 804d922c..6a5c5d1d 100644 --- a/plugins/opcua/plugin_handle.h +++ b/plugins/opcua/plugin_handle.h @@ -24,17 +24,47 @@ extern "C" { #endif +#include +#include + +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 } diff --git a/plugins/opcua/plugin_opcua.c b/plugins/opcua/plugin_opcua.c index 13ad989d..9ba1f1cf 100644 --- a/plugins/opcua/plugin_opcua.c +++ b/plugins/opcua/plugin_opcua.c @@ -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; diff --git a/plugins/sample/plugin_sample_drv.c b/plugins/sample/plugin_sample_drv.c index 3111590d..43ef9462 100644 --- a/plugins/sample/plugin_sample_drv.c +++ b/plugins/sample/plugin_sample_drv.c @@ -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; }