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

Update adapter loop to handle node subscribe/unsubscribe events

Browse Source
This commit is contained in:
eeff
2022-01-28 21:36:56 +08:00
committed by Parker Liu
parent 12e790575e
commit 30b20ee6ac
1 changed files with 41 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+41
View File
@@ -572,6 +572,38 @@ static int persister_singleton_handle_datatags(neu_adapter_t *adapter,
return rv;
}
static int persister_singleton_handle_subscriptions(neu_adapter_t *adapter,
neu_node_id_t node_id)
{
neu_persister_t *persister = persister_singleton_get();
char * adapter_name = NULL;
int rv = neu_manager_get_node_name_by_id(adapter->manager, node_id,
&adapter_name);
if (0 != rv) {
return rv;
}
vector_t *sub_infos = NULL;
rv = neu_manager_get_persist_subscription_infos(adapter->manager, node_id,
&sub_infos);
if (0 != rv) {
log_error("%s fail get persist subscription infos", adapter->name);
free(adapter_name);
return rv;
}
rv = neu_persister_store_subscriptions(persister, adapter_name, sub_infos);
if (0 != rv) {
log_error("%s fail store adapter:%s subscription infos", adapter_name,
adapter->name);
}
neu_persist_subscription_infos_free(sub_infos);
free(adapter_name);
return rv;
}
neu_plugin_running_state_e
neu_adapter_state_to_plugin_state(neu_adapter_t *adapter)
{
@@ -772,6 +804,15 @@ static void adapter_loop(void *arg)
break;
}
case MSG_EVENT_SUBSCRIBE_NODE:
// fall through
case MSG_EVENT_UNSUBSCRIBE_NODE: {
const neu_node_id_t *node_id_p = msg_get_buf_ptr(pay_msg);
persister_singleton_handle_subscriptions(adapter, *node_id_p);
break;
}
case MSG_EVENT_ADD_TAGS:
// fall through