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

Add forward subscribe/unsubscribe message to driver adapter

Browse Source 
* Add vector_find_index/vector_find_item/vector_has_elem util function
  for vector.
* Forward subscribe/unsubsribe message in manager, and make sure to
  forward the subscribe/unsubsribe message only once.
This commit is contained in:
Parker Liu
2021-09-22 18:21:37 +08:00
parent 4f5e858146
commit f552b41253
3 changed files with 218 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neu_types.h
+40
View File
@@ -45,6 +45,46 @@ typedef enum {
NEU_ORDER_MORE = 1
} neu_order_t;
/**
* Check if two value is equal.
*/
typedef bool (*neu_equal_func)(const void *a, const void *b);
/**
* Compare two value.
*/
typedef neu_order_t (*neu_compare_func)(const void *a, const void *b);
/**
* Check if the key is match the item
*/
typedef bool (*neu_key_match_func)(const void *key, const void *item);
/*
* define functions for implement equal function for primitive value
*/
#define prim_val_equal(locase_type, locase_type_t) \
static inline bool locase_type##_equal(const void *a, const void *b) \
{ \
return *(locase_type_t *) (a) == *(locase_type_t *) (b); \
}
#define prim_val_equal_t(locase_type) \
prim_val_equal(locase_type, locase_type##_t)
prim_val_equal_t(int8);
prim_val_equal_t(int16);
prim_val_equal_t(int32);
prim_val_equal_t(int64);
prim_val_equal_t(uint8);
prim_val_equal_t(uint16);
prim_val_equal_t(uint32);
prim_val_equal_t(uint64);
prim_val_equal(float, float);
prim_val_equal(double, double);
/**
* Boolean
* -------
include/neu_vector.h
+50
View File
@@ -26,9 +26,11 @@ extern "C" {
#include <stdlib.h>
#include "neu_types.h"
#include "vector.h"
typedef struct Vector vector_t;
typedef struct Vector neu_vector_t;
typedef struct Iterator iterator_t;
#define vector_init vector_setup
@@ -69,6 +71,54 @@ static inline vector_t *vector_clone(vector_t *vec)
return new_vec;
}
static inline size_t vector_find_index(vector_t *vec, const void *key,
neu_key_match_func match_func)
{
const void *item;
VECTOR_FOR_EACH(vec, iter)
{
item = (const void *) iterator_get(&iter);
if (match_func(key, item)) {
return iterator_index(vec, &iter);
}
}
return SIZE_MAX;
}
static inline void *vector_find_item(vector_t *vec, const void *key,
neu_key_match_func match_func)
{
const void *item;
VECTOR_FOR_EACH(vec, iter)
{
item = (const void *) iterator_get(&iter);
if (match_func(key, item)) {
return (void *) item;
}
}
return NULL;
}
static inline bool vector_has_elem(vector_t *vec, const void *elem,
neu_equal_func equ_func)
{
const void *item;
VECTOR_FOR_EACH(vec, iter)
{
item = (const void *) iterator_get(&iter);
if (equ_func(elem, item)) {
return true;
}
}
return false;
}
#ifdef __cplusplus
}
#endif
src/core/neu_manager.c
+128 -107
View File
@@ -348,59 +348,38 @@ static int uninit_bind_info(manager_bind_info_t *mng_bind_info)
return 0;
}
static bool match_id_reg_adapter(const void *key, const void *item)
{
return *(adapter_id_t *) key == ((adapter_reg_entity_t *) item)->adapter_id;
}
static bool match_name_reg_adapter(const void *key, const void *item)
{
neu_adapter_t *adapter;
adapter = ((adapter_reg_entity_t *) item)->adapter;
return strcmp((const char *) key, neu_adapter_get_name(adapter)) == 0;
}
// Return SIZE_MAX if can't find a adapter
static size_t find_reg_adapter_index_by_id(vector_t *adapters, adapter_id_t id)
{
size_t index = SIZE_MAX;
adapter_reg_entity_t *reg_entity;
VECTOR_FOR_EACH(adapters, iter)
{
reg_entity = (adapter_reg_entity_t *) iterator_get(&iter);
if (reg_entity->adapter_id == id) {
index = iterator_index(adapters, &iter);
break;
}
}
return index;
return vector_find_index(adapters, &id, match_id_reg_adapter);
}
static adapter_reg_entity_t *find_reg_adapter_by_id(vector_t * adapters,
adapter_id_t id)
{
adapter_reg_entity_t *reg_entity;
VECTOR_FOR_EACH(adapters, iter)
{
reg_entity = (adapter_reg_entity_t *) iterator_get(&iter);
if (reg_entity->adapter_id == id) {
return reg_entity;
}
}
return NULL;
return vector_find_item(adapters, &id, match_id_reg_adapter);
}
static adapter_reg_entity_t *find_reg_adapter_by_name(vector_t * adapters,
const char *name)
{
adapter_reg_entity_t *reg_entity;
const char * adapter_name;
VECTOR_FOR_EACH(adapters, iter)
{
reg_entity = (adapter_reg_entity_t *) iterator_get(&iter);
adapter_name = neu_adapter_get_name(reg_entity->adapter);
if (strcmp(adapter_name, name) == 0) {
return reg_entity;
}
}
return NULL;
return vector_find_item(adapters, name, match_name_reg_adapter);
}
static adapter_id_t manager_get_adapter_id(neu_manager_t *manager)
static adapter_id_t manager_new_adapter_id(neu_manager_t *manager)
{
adapter_id_t adapter_id;
@@ -512,7 +491,7 @@ static adapter_id_t manager_reg_adapter(neu_manager_t * manager,
return 0;
}
adapter_info.id = manager_get_adapter_id(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.plugin_id = plugin_reg_info.plugin_id;
@@ -725,32 +704,97 @@ static void stop_and_unreg_bind_adapters(neu_manager_t *manager)
}
}
static int add_grp_config_wait_bind(neu_manager_t * manager,
adapter_reg_entity_t *src_reg_entity,
adapter_reg_entity_t *dst_reg_entity,
neu_taggrp_config_t * grp_config)
static bool pipe_equal(const void *a, const void *b)
{
int rv = 0;
manager_bind_info_t *bind_info;
vector_t * sub_pipes;
return *(uint32_t *) a == *(uint32_t *) b;
}
bind_info = &manager->bind_info;
nng_mtx_lock(bind_info->mtx);
while (dst_reg_entity->bind_count == 0) {
nng_cv_wait(dst_reg_entity->cv);
}
nng_mtx_unlock(bind_info->mtx);
static int sub_grp_config_with_pipe(neu_taggrp_config_t *grp_config,
nng_pipe sub_pipe)
{
int rv = 0;
vector_t *sub_pipes;
sub_pipes = neu_taggrp_cfg_get_subpipes(grp_config);
// TODO: It's need to check if the adapter pipe is unique pipe in sub_pipes
rv = vector_push_back(sub_pipes, &dst_reg_entity->adapter_pipe);
if (vector_has_elem(sub_pipes, &sub_pipe, pipe_equal)) {
return 0;
}
rv = vector_push_back(sub_pipes, &sub_pipe);
if (rv != 0) {
log_error("Can't add pipe to vector of subscribe pipes");
return -1;
}
// if we add first one pipe, then need forward subscribe message
return sub_pipes->size == 1 ? 1 : 0;
}
static int sub_grp_config_with_adapter(neu_manager_t * manager,
neu_taggrp_config_t * grp_config,
adapter_reg_entity_t *reg_entity)
{
manager_bind_info_t *bind_info;
bind_info = &manager->bind_info;
nng_mtx_lock(bind_info->mtx);
// wait manage bind the adapter
while (reg_entity->bind_count == 0) {
nng_cv_wait(reg_entity->cv);
}
nng_mtx_unlock(bind_info->mtx);
return sub_grp_config_with_pipe(grp_config, reg_entity->adapter_pipe);
}
static int unsub_grp_config_with_pipe(neu_taggrp_config_t *grp_config,
nng_pipe sub_pipe)
{
int rv = 0;
size_t index;
vector_t *sub_pipes;
sub_pipes = neu_taggrp_cfg_get_subpipes(grp_config);
index = vector_find_index(sub_pipes, &sub_pipe, pipe_equal);
if (index == SIZE_MAX) {
return 0;
}
rv = vector_erase(sub_pipes, index);
if (rv != 0) {
log_error("Can't remove pipe from vector of subscribe pipes");
return -1;
}
// if we remove last one pipe, then need forward unsubscribe message
return sub_pipes->size == 0 ? 1 : 0;
}
static int unsub_grp_config_with_adapter(neu_manager_t * manager,
neu_taggrp_config_t * grp_config,
adapter_reg_entity_t *reg_entity)
{
manager_bind_info_t *bind_info;
bind_info = &manager->bind_info;
nng_mtx_lock(bind_info->mtx);
// wait manage bind the adapter
while (reg_entity->bind_count == 0) {
nng_cv_wait(reg_entity->cv);
}
nng_mtx_unlock(bind_info->mtx);
return unsub_grp_config_with_pipe(grp_config, reg_entity->adapter_pipe);
}
static int add_grp_config_to_adapter(adapter_reg_entity_t *reg_entity,
neu_taggrp_config_t * grp_config)
{
int rv = 0;
neu_datatag_manager_t *datatag_mng;
datatag_mng = src_reg_entity->datatag_manager;
datatag_mng = reg_entity->datatag_manager;
rv = neu_datatag_mng_add_grp_config(datatag_mng, grp_config);
if (rv != 0) {
log_error("Failed to add datatag group config: %s",
@@ -765,22 +809,18 @@ static int manager_add_config_by_name(neu_manager_t * manager,
config_add_cmd_t *cmd)
{
int rv = 0;
adapter_reg_entity_t *src_reg_entity;
adapter_reg_entity_t *dst_reg_entity;
adapter_reg_entity_t *reg_entity;
nng_mtx_lock(manager->adapters_mtx);
src_reg_entity =
reg_entity =
find_reg_adapter_by_name(&manager->reg_adapters, cmd->src_adapter_name);
dst_reg_entity =
find_reg_adapter_by_name(&manager->reg_adapters, cmd->dst_adapter_name);
if (src_reg_entity == NULL || dst_reg_entity == NULL) {
log_error("Can't find matched src or dst registered adapter");
if (reg_entity == NULL) {
log_error("Can't find matched src registered adapter");
rv = -1;
goto add_config_by_name_exit;
}
rv = add_grp_config_wait_bind(manager, src_reg_entity, dst_reg_entity,
cmd->grp_config);
rv = add_grp_config_to_adapter(reg_entity, cmd->grp_config);
add_config_by_name_exit:
nng_mtx_unlock(manager->adapters_mtx);
return rv;
@@ -1016,6 +1056,7 @@ static void manager_loop(void *arg)
break;
}
/* clang-format off
case MSG_CMD_START_PERIODIC_READ: {
size_t msg_size;
nng_msg * out_msg;
@@ -1089,11 +1130,13 @@ static void manager_loop(void *arg)
}
break;
}
clang-format on */
case MSG_CMD_SUBSCRIBE_NODE: {
size_t msg_size;
nng_msg * out_msg;
nng_pipe msg_pipe;
int need_forward;
adapter_id_t adapter_id;
subscribe_node_cmd_t *cmd_ptr;
@@ -1109,7 +1152,9 @@ static void manager_loop(void *arg)
nng_mtx_unlock(manager->adapters_mtx);
msg_size = msg_inplace_data_get_size(sizeof(subscribe_node_cmd_t));
rv = nng_msg_alloc(&out_msg, msg_size);
if (rv == 0) {
need_forward = sub_grp_config_with_pipe(cmd_ptr->grp_config,
nng_msg_get_pipe(msg));
if (rv == 0 && need_forward == 1) {
message_t * msg_ptr;
subscribe_node_cmd_t *out_cmd_ptr;
msg_ptr = (message_t *) nng_msg_body(out_msg);
@@ -1129,6 +1174,7 @@ static void manager_loop(void *arg)
size_t msg_size;
nng_msg * out_msg;
nng_pipe msg_pipe;
int need_forward;
adapter_id_t adapter_id;
unsubscribe_node_cmd_t *cmd_ptr;
@@ -1144,8 +1190,10 @@ static void manager_loop(void *arg)
nng_mtx_unlock(manager->adapters_mtx);
msg_size =
msg_inplace_data_get_size(sizeof(unsubscribe_node_cmd_t));
rv = nng_msg_alloc(&out_msg, msg_size);
if (rv == 0) {
rv = nng_msg_alloc(&out_msg, msg_size);
need_forward = unsub_grp_config_with_pipe(cmd_ptr->grp_config,
nng_msg_get_pipe(msg));
if (rv == 0 || need_forward == 1) {
message_t * msg_ptr;
unsubscribe_node_cmd_t *out_cmd_ptr;
msg_ptr = (message_t *) nng_msg_body(out_msg);
@@ -1436,10 +1484,8 @@ int neu_manager_add_grp_config(neu_manager_t * manager,
neu_cmd_add_grp_config_t *cmd)
{
int rv = 0;
adapter_id_t src_adapter_id;
adapter_id_t dst_adapter_id;
adapter_reg_entity_t *src_reg_entity;
adapter_reg_entity_t *dst_reg_entity;
adapter_id_t adapter_id;
adapter_reg_entity_t *reg_entity;
if (manager == NULL || cmd == NULL) {
log_error("Add group config with NULL manager or command");
@@ -1452,22 +1498,15 @@ int neu_manager_add_grp_config(neu_manager_t * manager,
}
nng_mtx_lock(manager->adapters_mtx);
src_adapter_id =
neu_manager_adapter_id_from_node_id(manager, cmd->src_node_id);
dst_adapter_id =
neu_manager_adapter_id_from_node_id(manager, cmd->dst_node_id);
src_reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, src_adapter_id);
dst_reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, dst_adapter_id);
if (src_reg_entity == NULL || dst_reg_entity == NULL) {
log_error("Can't find matched src or dst registered adapter");
adapter_id = neu_manager_adapter_id_from_node_id(manager, cmd->src_node_id);
reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id);
if (reg_entity == NULL) {
log_error("Can't find matched src registered adapter");
rv = -1;
goto add_grp_config_exit;
}
rv = add_grp_config_wait_bind(manager, src_reg_entity, dst_reg_entity,
cmd->grp_config);
rv = add_grp_config_to_adapter(reg_entity, cmd->grp_config);
add_grp_config_exit:
nng_mtx_unlock(manager->adapters_mtx);
return rv;
@@ -1508,10 +1547,8 @@ int neu_manager_update_grp_config(neu_manager_t * manager,
neu_cmd_update_grp_config_t *cmd)
{
int rv = 0;
adapter_id_t src_adapter_id;
adapter_id_t dst_adapter_id;
adapter_reg_entity_t *src_reg_entity;
adapter_reg_entity_t *dst_reg_entity;
adapter_id_t adapter_id;
adapter_reg_entity_t *reg_entity;
if (manager == NULL || cmd == NULL) {
log_error("Update group config with NULL manager or command");
@@ -1524,31 +1561,15 @@ int neu_manager_update_grp_config(neu_manager_t * manager,
}
nng_mtx_lock(manager->adapters_mtx);
src_adapter_id =
neu_manager_adapter_id_from_node_id(manager, cmd->src_node_id);
dst_adapter_id =
neu_manager_adapter_id_from_node_id(manager, cmd->dst_node_id);
src_reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, src_adapter_id);
dst_reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, dst_adapter_id);
if (src_reg_entity == NULL || dst_reg_entity == NULL) {
adapter_id = neu_manager_adapter_id_from_node_id(manager, cmd->src_node_id);
reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id);
if (reg_entity == NULL) {
log_error("Can't find matched src or dst registered adapter");
rv = -1;
goto update_grp_config_exit;
}
vector_t *sub_pipes;
sub_pipes = neu_taggrp_cfg_get_subpipes(cmd->grp_config);
// TODO: It's need to check if the adapter pipe is unique pipe in sub_pipes
rv = vector_push_back(sub_pipes, &dst_reg_entity->adapter_pipe);
if (rv != 0) {
log_error("Can't add pipe to vector of subscribe pipes");
rv = -1;
goto update_grp_config_exit;
}
rv = neu_datatag_mng_update_grp_config(src_reg_entity->datatag_manager,
rv = neu_datatag_mng_update_grp_config(reg_entity->datatag_manager,
cmd->grp_config);
if (rv != 0) {
log_error("Failed to update datatag group config: %s",