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

feat(gw-sn): call subscribe/unsubscribe hook

This commit is contained in:
JianBo He 2021-07-30 14:49:54 +08:00
parent c986f89319
commit 602f0ebb60
2 changed files with 44 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
apps/emqx_gateway/src/mqttsn/emqx_sn_channel.erl
View File

@ -845,12 +845,23 @@ check_subscribe_authz({_TopicId, TopicName, _QoS},
do_subscribe({TopicId, TopicName, QoS}, do_subscribe({TopicId, TopicName, QoS},
Channel = #channel{ Channel = #channel{
ctx = Ctx,
session = Session, session = Session,
clientinfo = ClientInfo clientinfo = ClientInfo
= #{mountpoint := Mountpoint}}) -> = #{mountpoint := Mountpoint}}) ->
NTopicName = emqx_mountpoint:mount(Mountpoint, TopicName),
SubOpts = maps:merge(?DEFAULT_SUBOPTS, #{qos => QoS}), {TopicName1, SubOpts0} = emqx_topic:parse(TopicName),
case emqx_session:subscribe(ClientInfo, NTopicName, SubOpts, Session) of TopicFilters = [{TopicName1, SubOpts0#{qos => QoS}}],
case run_hooks(Ctx, 'client.subscribe',
[ClientInfo, #{}], TopicFilters) of
[] ->
?LOG(warning, "Skip to subscribe ~s, "
"due to 'client.subscribe' denied!", [TopicName]),
{ok, Channel};
[{NTopicName, NSubOpts}|_] ->
NTopicName1 = emqx_mountpoint:mount(Mountpoint, NTopicName),
NSubOpts1 = maps:merge(?DEFAULT_SUBOPTS, NSubOpts),
case emqx_session:subscribe(ClientInfo, NTopicName1, NSubOpts1, Session) of
{ok, NSession} -> {ok, NSession} ->
{ok, {TopicId, QoS}, {ok, {TopicId, QoS},
Channel#channel{session = NSession}}; Channel#channel{session = NSession}};
@ -858,6 +869,7 @@ do_subscribe({TopicId, TopicName, QoS},
?LOG(warning, "Cannot subscribe ~s due to ~s.", ?LOG(warning, "Cannot subscribe ~s due to ~s.",
[TopicName, emqx_reason_codes:text(?RC_QUOTA_EXCEEDED)]), [TopicName, emqx_reason_codes:text(?RC_QUOTA_EXCEEDED)]),
{error, ?SN_EXCEED_LIMITATION} {error, ?SN_EXCEED_LIMITATION}
end
end. end.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
@ -891,16 +903,29 @@ preproc_unsub_type(?SN_UNSUBSCRIBE_MSG_TYPE(?SN_SHORT_TOPIC,
do_unsubscribe(TopicName, do_unsubscribe(TopicName,
Channel = #channel{ Channel = #channel{
ctx = Ctx,
session = Session, session = Session,
clientinfo = ClientInfo clientinfo = ClientInfo
= #{mountpoint := Mountpoint}}) -> = #{mountpoint := Mountpoint}}) ->
NTopicName = emqx_mountpoint:mount(Mountpoint, TopicName), TopicFilters = [emqx_topic:parse(TopicName)],
case emqx_session:unsubscribe(ClientInfo, NTopicName, case run_hooks(Ctx, 'client.unsubscribe',
?DEFAULT_SUBOPTS, Session) of [ClientInfo, #{}], TopicFilters) of
[] ->
%% Skip to unsubscribe
{ok, Channel};
[{NTopicName, NSubOpts}|_] ->
NTopicName1 = emqx_mountpoint:mount(Mountpoint, NTopicName),
NSubOpts1 = maps:merge(
emqx_gateway_utils:default_subopts(),
NSubOpts
),
case emqx_session:unsubscribe(ClientInfo, NTopicName1,
NSubOpts1, Session) of
{ok, NSession} -> {ok, NSession} ->
{ok, Channel#channel{session = NSession}}; {ok, Channel#channel{session = NSession}};
{error, ?RC_NO_SUBSCRIPTION_EXISTED} -> {error, ?RC_NO_SUBSCRIPTION_EXISTED} ->
{ok, Channel} {ok, Channel}
end
end. end.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------

3
apps/emqx_gateway/src/mqttsn/emqx_sn_registry.erl
View File

@ -155,7 +155,8 @@ init([InstaId, PredefTopics]) ->
% FIXME: % FIXME:
%ok = ekka_rlog:wait_for_shards([?CM_SHARD], infinity), %ok = ekka_rlog:wait_for_shards([?CM_SHARD], infinity),
MaxPredefId = lists:foldl( MaxPredefId = lists:foldl(
fun(#{id := TopicId, topic := TopicName}, AccId) -> fun(#{id := TopicId, topic := TopicName0}, AccId) ->
TopicName = iolist_to_binary(TopicName0),
ekka_mnesia:dirty_write(Tab, #emqx_sn_registry{ ekka_mnesia:dirty_write(Tab, #emqx_sn_registry{
key = {predef, TopicId}, key = {predef, TopicId},
value = TopicName} value = TopicName}