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

Add send subscribe node message when Add default group configs

Browse Source
This commit is contained in:
Parker Liu
2021-09-26 17:15:59 +08:00
parent ba89625545
commit 3b369acf25
5 changed files with 67 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/modbus/modbus_tcp.c
+2
View File
@@ -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);
plugins/opcua/plugin_opcua.c
+1
View File
@@ -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);
plugins/sample/plugin_sample_app.c
+7 -4
View File
@@ -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;
}
plugins/sample/plugin_sample_drv.c
+3 -3
View File
@@ -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;
src/core/neu_manager.c
+54 -17
View File
@@ -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;
}