From 3b369acf256cb19e1f495c95f2cf146824b07be5 Mon Sep 17 00:00:00 2001 From: Parker Liu Date: Sun, 26 Sep 2021 17:15:59 +0800 Subject: [PATCH] Add send subscribe node message when Add default group configs --- plugins/modbus/modbus_tcp.c | 2 + plugins/opcua/plugin_opcua.c | 1 + plugins/sample/plugin_sample_app.c | 11 +++-- plugins/sample/plugin_sample_drv.c | 6 +-- src/core/neu_manager.c | 71 +++++++++++++++++++++++------- 5 files changed, 67 insertions(+), 24 deletions(-) diff --git a/plugins/modbus/modbus_tcp.c b/plugins/modbus/modbus_tcp.c index c1208736..75e9ddbe 100644 --- a/plugins/modbus/modbus_tcp.c +++ b/plugins/modbus/modbus_tcp.c @@ -118,6 +118,8 @@ static void start_periodic_read(neu_plugin_t * plugin, modbus_point_new_cmd(sub_inst->point_ctx); + log_info("pre start periodic read, grp: %s(%p), interval: %d", + sub_inst->name, grp_config, interval); pthread_mutex_lock(&plugin->mtx); TAILQ_INSERT_TAIL(&plugin->sub_instances, sub_inst, node); pthread_mutex_unlock(&plugin->mtx); diff --git a/plugins/opcua/plugin_opcua.c b/plugins/opcua/plugin_opcua.c index 47b01dc4..2f48013b 100644 --- a/plugins/opcua/plugin_opcua.c +++ b/plugins/opcua/plugin_opcua.c @@ -169,6 +169,7 @@ static int opcua_plugin_request(neu_plugin_t *plugin, neu_request_t *req) return -1; } + log_info("plugin context is %p", plugin->handle_context); if (NULL == plugin->handle_context->table) { plugin->handle_context->table = neu_system_get_datatags_table( plugin, plugin->handle_context->self_node_id); diff --git a/plugins/sample/plugin_sample_app.c b/plugins/sample/plugin_sample_app.c index e46ea389..072f2936 100644 --- a/plugins/sample/plugin_sample_app.c +++ b/plugins/sample/plugin_sample_app.c @@ -387,19 +387,22 @@ static int handle_trans_data_value(neu_data_val_t *trans_val) neu_data_val_t *val_i64; neu_data_val_t *val_f64; neu_data_val_t *val_cstr; - int64_t tag001; - double tag002; + int64_t i64; + double f64; char * cstr; int_val = neu_fixed_array_get(array, 0); val_i64 = int_val->val; - neu_dvalue_get_int64(val_i64, &tag001); + neu_dvalue_get_int64(val_i64, &i64); int_val = neu_fixed_array_get(array, 1); val_f64 = int_val->val; - neu_dvalue_get_double(val_f64, &tag002); + neu_dvalue_get_double(val_f64, &f64); int_val = neu_fixed_array_get(array, 2); val_cstr = int_val->val; neu_dvalue_get_ref_cstr(val_cstr, &cstr); + + log_info("The sample driver read data, i64: %d, f64: %f, cstr: %s", i64, + f64, cstr); return 0; } diff --git a/plugins/sample/plugin_sample_drv.c b/plugins/sample/plugin_sample_drv.c index d4958ba1..7c1b1cf2 100644 --- a/plugins/sample/plugin_sample_drv.c +++ b/plugins/sample/plugin_sample_drv.c @@ -26,9 +26,9 @@ const neu_plugin_module_t neu_plugin_module; #define TAG_DATA_STR_LEN 100 -static int64_t tag001; -static double tag002; -static char tag003[TAG_DATA_STR_LEN + 1]; +static int64_t tag001 = 42; +static double tag002 = 2.718281828; +static char tag003[TAG_DATA_STR_LEN + 1] = "Hello, sample app"; struct neu_plugin { neu_plugin_common_t common; diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index ee63def7..6b40acc5 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -258,7 +258,7 @@ static const plugin_reg_param_t default_plugin_infos[] = { typedef struct config_add_cmd { char * config_name; char * src_adapter_name; - char * dst_adapter_name; + char * sub_adapter_name; uint32_t read_interval; neu_taggrp_config_t *grp_config; } config_add_cmd_t; @@ -268,42 +268,42 @@ static config_add_cmd_t default_config_add_cmds[] = { { .config_name = "config_drv_sample", .src_adapter_name = SAMPLE_DRV_ADAPTER_NAME, - .dst_adapter_name = SAMPLE_APP_ADAPTER_NAME, + .sub_adapter_name = SAMPLE_APP_ADAPTER_NAME, .read_interval = 2000, .grp_config = NULL, }, { .config_name = "config_app_sample", .src_adapter_name = SAMPLE_APP_ADAPTER_NAME, - .dst_adapter_name = SAMPLE_DRV_ADAPTER_NAME, + .sub_adapter_name = SAMPLE_DRV_ADAPTER_NAME, .read_interval = 2000, .grp_config = NULL, }, { .config_name = "config_mqtt_sample", .src_adapter_name = SAMPLE_DRV_ADAPTER_NAME, - .dst_adapter_name = MQTT_ADAPTER_NAME, + .sub_adapter_name = MQTT_ADAPTER_NAME, .read_interval = 2000, .grp_config = NULL, }, { .config_name = "config_modbus_tcp_sample", .src_adapter_name = SAMPLE_APP_ADAPTER_NAME, - .dst_adapter_name = MODBUS_TCP_ADAPTER_NAME, + .sub_adapter_name = MODBUS_TCP_ADAPTER_NAME, .read_interval = 2000, .grp_config = NULL, }, { .config_name = "config_modbus_tcp_sample_2", .src_adapter_name = MODBUS_TCP_ADAPTER_NAME, - .dst_adapter_name = MQTT_ADAPTER_NAME, + .sub_adapter_name = MQTT_ADAPTER_NAME, .read_interval = 2000, .grp_config = NULL, }, { .config_name = "config_opcua_sample", .src_adapter_name = OPCUA_ADAPTER_NAME, - .dst_adapter_name = MQTT_ADAPTER_NAME, + .sub_adapter_name = MQTT_ADAPTER_NAME, .read_interval = 2000, .grp_config = NULL, }, @@ -833,17 +833,47 @@ static void add_default_grp_configs(neu_manager_t *manager) uint32_t i; config_add_cmd_t * config_add_cmd; neu_taggrp_config_t * grp_config; - adapter_reg_entity_t *reg_entity; + adapter_reg_entity_t *src_reg_entity; + adapter_reg_entity_t *sub_reg_entity; + + size_t msg_size; + nng_msg *out_msg; + int need_send; + int rv; for (i = 0; i < DEFAULT_GROUP_CONFIG_COUNT; i++) { config_add_cmd = &default_config_add_cmds[i]; grp_config = neu_taggrp_cfg_new(config_add_cmd->config_name); neu_taggrp_cfg_set_interval(grp_config, config_add_cmd->read_interval); config_add_cmd->grp_config = grp_config; - reg_entity = find_reg_adapter_by_name(&manager->reg_adapters, - config_add_cmd->dst_adapter_name); - add_grp_config_to_adapter(reg_entity, grp_config); - sub_grp_config_with_adapter(manager, grp_config, reg_entity); + src_reg_entity = find_reg_adapter_by_name( + &manager->reg_adapters, config_add_cmd->src_adapter_name); + sub_reg_entity = find_reg_adapter_by_name( + &manager->reg_adapters, config_add_cmd->sub_adapter_name); + add_grp_config_to_adapter(src_reg_entity, grp_config); + need_send = + sub_grp_config_with_adapter(manager, grp_config, sub_reg_entity); + + /* Send the subscribe node message to driver + */ + msg_size = msg_inplace_data_get_size(sizeof(subscribe_node_cmd_t)); + rv = nng_msg_alloc(&out_msg, msg_size); + if (rv == 0 && need_send == 1) { + message_t * msg_ptr; + subscribe_node_cmd_t *out_cmd_ptr; + msg_ptr = (message_t *) nng_msg_body(out_msg); + msg_inplace_data_init(msg_ptr, MSG_CMD_SUBSCRIBE_NODE, + sizeof(subscribe_node_cmd_t)); + out_cmd_ptr = msg_get_buf_ptr(msg_ptr); + out_cmd_ptr->grp_config = grp_config; + out_cmd_ptr->sender_id = 0; + out_cmd_ptr->dst_node_id = neu_manager_adapter_id_to_node_id( + manager, src_reg_entity->adapter_id); + nng_msg_set_pipe(out_msg, src_reg_entity->adapter_pipe); + log_info("Send subscribe driver command to driver pipe: %d", + src_reg_entity->adapter_pipe); + nng_sendmsg(manager->bind_info.mng_sock, out_msg, 0); + } } return; } @@ -853,19 +883,26 @@ static void remove_default_grp_configs(neu_manager_t *manager) uint32_t i; config_add_cmd_t * config_add_cmd; neu_taggrp_config_t * grp_config; - adapter_reg_entity_t *reg_entity; + adapter_reg_entity_t *src_reg_entity; + adapter_reg_entity_t *sub_reg_entity; for (i = 0; i < DEFAULT_GROUP_CONFIG_COUNT; i++) { config_add_cmd = &default_config_add_cmds[i]; grp_config = config_add_cmd->grp_config; nng_mtx_lock(manager->adapters_mtx); - reg_entity = find_reg_adapter_by_name(&manager->reg_adapters, - config_add_cmd->dst_adapter_name); - unsub_grp_config_with_adapter(manager, grp_config, reg_entity); - neu_datatag_mng_del_grp_config(reg_entity->datatag_manager, + src_reg_entity = find_reg_adapter_by_name( + &manager->reg_adapters, config_add_cmd->src_adapter_name); + sub_reg_entity = find_reg_adapter_by_name( + &manager->reg_adapters, config_add_cmd->sub_adapter_name); + unsub_grp_config_with_adapter(manager, grp_config, sub_reg_entity); + neu_datatag_mng_del_grp_config(src_reg_entity->datatag_manager, config_add_cmd->config_name); nng_mtx_unlock(manager->adapters_mtx); + + /* The neuron manager should be shutdown, so there is no need to + * send the unsubscribe node message to driver. + */ } return; }