Use event-based mechanism for stream discovery #14130
Conversation
ieQu1
force-pushed
the
dev/new-stream-events
branch
2 times, most recently
from
3a468d2 to
91685aa
Compare
ieQu1
changed the title
ieQu1
force-pushed
the
dev/new-stream-events
branch
3 times, most recently
from
1194284 to
bcbeb9f
Compare
ieQu1
force-pushed
the
dev/new-stream-events
branch
from
bcbeb9f to
8a7d56b
Compare
apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_subs.erl
Outdated
Show resolved
Hide resolved
apps/emqx/src/emqx_persistent_session_ds/emqx_persistent_session_ds_subs.erl
Outdated
Show resolved
Hide resolved
| %% all nodes. | ||
| -spec notify_new_stream(emqx_ds:db(), emqx_ds:topic_filter()) -> ok. | ||
| notify_new_stream(DB, TF) -> | ||
| emqx_ds_new_streams_proto_v1:notify([node() | nodes()], DB, TF). |
There was a problem hiding this comment.
Don't we need running_nodes here?
There was a problem hiding this comment.
In theory, yes. But I don't like the kludge of using mria:running_nodes() for unintended purpose.
It's mostly used as a workaround to avoid sending rpc calls to nodes that can't handle them (e.g. rshell), it's less relevant for casts, since reply is not expected.
| %% "active" flag. Whenever it receives a notification about a new | ||
| %% stream, it matches all active subscriptions' topic-filters against | ||
| %% the topic-filter of the event, and sets `active' flags for every | ||
| %% match. |
There was a problem hiding this comment.
Probably some explantion of "dirty" therminology is needed.
|
|
||
| register_db(DB, Backend) -> | ||
| ets:insert(?TAB, {DB, Backend}), | ||
| _ = supervisor:start_child(?WATCH_SUP, [DB]), |
There was a problem hiding this comment.
How is this supervisor stopped on db close/drop?
There was a problem hiding this comment.
It's never stopped by design. Stopping it would lose subscriptions and all events. I'll add a comment.
|
|
||
| %% @doc This module is used for notifying processes about new streams. | ||
| %% It's not a replacement for `emqx_ds:get_streams' function, it's | ||
| %% only meant to optimize its usage. |
There was a problem hiding this comment.
The relation of the API that this module provides to the DS API may be still unclear: is it available for all DS'es? Which DS'es should implement it? How is it possible for a DS consumer understand whether it is possible to use it?
There was a problem hiding this comment.
Right now the idea is that any DS backend/layout can use this module at its own discretion. I didn't have time to design a fancy layered API.
Fixes
Release version: v/e5.?
Summary
PR Checklist
Please convert it to a draft if any of the following conditions are not met. Reviewers may skip over until all the items are checked:
changes/(ce|ee)/(feat|perf|fix|breaking)-<PR-id>.en.mdfilesChecklist for CI (.github/workflows) changes
changes/dir for user-facing artifacts update