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

refactor(exclusive_sub): Export transaction

This commit is contained in:
ieQu1 2022-08-18 12:19:34 +02:00
parent b962a71da2
commit 4fa06fcd63
1 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
apps/emqx/src/emqx_exclusive_subscription.erl
View File

@ -35,6 +35,11 @@
unsubscribe/2 unsubscribe/2
]). ]).
%% Internal exports (RPC)
-export([
try_subscribe/2
]).
-record(exclusive_subscription, { -record(exclusive_subscription, {
topic :: emqx_types:topic(), topic :: emqx_types:topic(),
clientid :: emqx_types:clientid() clientid :: emqx_types:clientid()
@ -80,10 +85,7 @@ on_delete_module() ->
-spec check_subscribe(emqx_types:clientinfo(), emqx_types:topic()) -> -spec check_subscribe(emqx_types:clientinfo(), emqx_types:topic()) ->
allow | deny. allow | deny.
check_subscribe(#{clientid := ClientId}, Topic) -> check_subscribe(#{clientid := ClientId}, Topic) ->
Fun = fun() -> case mria:transaction(?EXCLUSIVE_SHARD, fun ?MODULE:try_subscribe/2, [ClientId, Topic]) of
try_subscribe(ClientId, Topic)
end,
case mria:transaction(?EXCLUSIVE_SHARD, Fun) of
{atomic, Res} -> {atomic, Res} ->
Res; Res;
{aborted, Reason} -> {aborted, Reason} ->
@ -94,7 +96,7 @@ check_subscribe(#{clientid := ClientId}, Topic) ->
end. end.
unsubscribe(Topic, #{is_exclusive := true}) -> unsubscribe(Topic, #{is_exclusive := true}) ->
_ = mria:transaction(?EXCLUSIVE_SHARD, fun() -> mnesia:delete({?TAB, Topic}) end), _ = mria:transaction(?EXCLUSIVE_SHARD, fun mnesia:delete/1, [{?TAB, Topic}]),
ok; ok;
unsubscribe(_Topic, _SubOpts) -> unsubscribe(_Topic, _SubOpts) ->
ok. ok.