yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: ensure iterator is opened

This commit is contained in:
Thales Macedo Garitezi 2023-08-11 10:06:39 -03:00
parent 33ddbe80ad
commit c28c6d1b7e
1 changed files with 13 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
apps/emqx/src/emqx_persistent_session_ds.erl
View File

@ -93,12 +93,7 @@ add_subscription(TopicFilterBin, DSSessionID) ->
{ok, IteratorID, StartMS, IsNew} = emqx_ds:session_add_iterator( {ok, IteratorID, StartMS, IsNew} = emqx_ds:session_add_iterator(
DSSessionID, TopicFilter DSSessionID, TopicFilter
), ),
case IsNew of ok = open_iterator_on_all_nodes(TopicFilter, StartMS, IteratorID),
true ->
ok = open_iterator_on_all_nodes(TopicFilter, StartMS, IteratorID);
false ->
ok
end,
{ok, IteratorID, IsNew} {ok, IteratorID, IsNew}
end end
). ).
@ -106,7 +101,9 @@ add_subscription(TopicFilterBin, DSSessionID) ->
-spec open_iterator_on_all_nodes(emqx_topic:words(), emqx_ds:time(), emqx_ds:iterator_id()) -> ok. -spec open_iterator_on_all_nodes(emqx_topic:words(), emqx_ds:time(), emqx_ds:iterator_id()) -> ok.
open_iterator_on_all_nodes(TopicFilter, StartMS, IteratorID) -> open_iterator_on_all_nodes(TopicFilter, StartMS, IteratorID) ->
Nodes = emqx:running_nodes(), Nodes = emqx:running_nodes(),
Results = emqx_persistent_session_ds_proto_v1:open_iterator(Nodes, TopicFilter, StartMS, IteratorID), Results = emqx_persistent_session_ds_proto_v1:open_iterator(
Nodes, TopicFilter, StartMS, IteratorID
),
%% TODO: handle errors %% TODO: handle errors
true = lists:all(fun(Res) -> Res =:= {ok, ok} end, Results), true = lists:all(fun(Res) -> Res =:= {ok, ok} end, Results),
ok. ok.
@ -114,10 +111,15 @@ open_iterator_on_all_nodes(TopicFilter, StartMS, IteratorID) ->
-spec do_open_iterator(emqx_topic:words(), emqx_ds:time(), emqx_ds:iterator_id()) -> ok. -spec do_open_iterator(emqx_topic:words(), emqx_ds:time(), emqx_ds:iterator_id()) -> ok.
do_open_iterator(TopicFilter, StartMS, IteratorID) -> do_open_iterator(TopicFilter, StartMS, IteratorID) ->
Replay = {TopicFilter, StartMS}, Replay = {TopicFilter, StartMS},
%% FIXME: choose DS shard based on ...? case emqx_ds_storage_layer:is_iterator_present(?DS_SHARD, IteratorID) of
{ok, It} = emqx_ds_storage_layer:make_iterator(?DS_SHARD, Replay), true ->
ok = emqx_ds_storage_layer:preserve_iterator(It, IteratorID), {ok, _It} = emqx_ds_storage_layer:restore_iterator(?DS_SHARD, IteratorID),
ok. ok;
false ->
{ok, It} = emqx_ds_storage_layer:make_iterator(?DS_SHARD, Replay),
ok = emqx_ds_storage_layer:preserve_iterator(It, IteratorID),
ok
end.
%% %%