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

fix(mgmt): add subscribe options in client subscribe API

This commit is contained in:
firest 2022-04-27 12:42:22 +08:00
parent 8d01e8a697
commit d9c3cf5c97
2 changed files with 25 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
apps/emqx_management/src/emqx_mgmt_api_clients.erl
View File

@ -513,7 +513,10 @@ fields(keepalive) ->
fields(subscribe) -> fields(subscribe) ->
[ [
{topic, hoconsc:mk(binary(), #{desc => <<"Topic">>})}, {topic, hoconsc:mk(binary(), #{desc => <<"Topic">>})},
{qos, hoconsc:mk(emqx_schema:qos(), #{desc => <<"QoS">>})} {qos, hoconsc:mk(emqx_schema:qos(), #{desc => <<"QoS">>})},
{nl, hoconsc:mk(integer(), #{default => 0, desc => <<"No Local">>})},
{rap, hoconsc:mk(integer(), #{default => 0, desc => <<"Retain as Published">>})},
{rh, hoconsc:mk(integer(), #{default => 0, desc => <<"Retain Handling">>})}
]; ];
fields(unsubscribe) -> fields(unsubscribe) ->
[ [
@ -536,9 +539,8 @@ authz_cache(delete, #{bindings := Bindings}) ->
clean_authz_cache(Bindings). clean_authz_cache(Bindings).
subscribe(post, #{bindings := #{clientid := ClientID}, body := TopicInfo}) -> subscribe(post, #{bindings := #{clientid := ClientID}, body := TopicInfo}) ->
Topic = maps:get(<<"topic">>, TopicInfo), Opts = emqx_map_lib:unsafe_atom_key_map(TopicInfo),
Qos = maps:get(<<"qos">>, TopicInfo, 0), subscribe(Opts#{clientid => ClientID}).
subscribe(#{clientid => ClientID, topic => Topic, qos => Qos}).
unsubscribe(post, #{bindings := #{clientid := ClientID}, body := TopicInfo}) -> unsubscribe(post, #{bindings := #{clientid := ClientID}, body := TopicInfo}) ->
Topic = maps:get(<<"topic">>, TopicInfo), Topic = maps:get(<<"topic">>, TopicInfo),
@ -548,11 +550,7 @@ unsubscribe(post, #{bindings := #{clientid := ClientID}, body := TopicInfo}) ->
subscribe_batch(post, #{bindings := #{clientid := ClientID}, body := TopicInfos}) -> subscribe_batch(post, #{bindings := #{clientid := ClientID}, body := TopicInfos}) ->
Topics = Topics =
[ [
begin emqx_map_lib:unsafe_atom_key_map(TopicInfo)
Topic = maps:get(<<"topic">>, TopicInfo),
Qos = maps:get(<<"qos">>, TopicInfo, 0),
#{topic => Topic, qos => Qos}
end
|| TopicInfo <- TopicInfos || TopicInfo <- TopicInfos
], ],
subscribe_batch(#{clientid => ClientID, topics => Topics}). subscribe_batch(#{clientid => ClientID, topics => Topics}).
@ -661,21 +659,16 @@ clean_authz_cache(#{clientid := ClientID}) ->
{500, #{code => <<"UNKNOW_ERROR">>, message => Message}} {500, #{code => <<"UNKNOW_ERROR">>, message => Message}}
end. end.
subscribe(#{clientid := ClientID, topic := Topic, qos := Qos}) -> subscribe(#{clientid := ClientID, topic := Topic} = Sub) ->
case do_subscribe(ClientID, Topic, Qos) of Opts = maps:with([qos, nl, rap, rh], Sub),
case do_subscribe(ClientID, Topic, Opts) of
{error, channel_not_found} -> {error, channel_not_found} ->
{404, ?CLIENT_ID_NOT_FOUND}; {404, ?CLIENT_ID_NOT_FOUND};
{error, Reason} -> {error, Reason} ->
Message = list_to_binary(io_lib:format("~p", [Reason])), Message = list_to_binary(io_lib:format("~p", [Reason])),
{500, #{code => <<"UNKNOW_ERROR">>, message => Message}}; {500, #{code => <<"UNKNOW_ERROR">>, message => Message}};
{ok, Node} -> {ok, Node} ->
Response = Response = Sub#{node => Node},
#{
clientid => ClientID,
topic => Topic,
qos => Qos,
node => Node
},
{200, Response} {200, Response}
end. end.
@ -688,15 +681,18 @@ unsubscribe(#{clientid := ClientID, topic := Topic}) ->
end. end.
subscribe_batch(#{clientid := ClientID, topics := Topics}) -> subscribe_batch(#{clientid := ClientID, topics := Topics}) ->
ArgList = [[ClientID, Topic, Qos] || #{topic := Topic, qos := Qos} <- Topics], ArgList = [
[ClientID, Topic, maps:with([qos, nl, rap, rh], Sub)]
|| #{topic := Topic} = Sub <- Topics
],
emqx_mgmt_util:batch_operation(?MODULE, do_subscribe, ArgList). emqx_mgmt_util:batch_operation(?MODULE, do_subscribe, ArgList).
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% internal function %% internal function
do_subscribe(ClientID, Topic0, Qos) -> do_subscribe(ClientID, Topic0, Options) ->
{Topic, Opts} = emqx_topic:parse(Topic0), {Topic, Opts} = emqx_topic:parse(Topic0, Options),
TopicTable = [{Topic, Opts#{qos => Qos}}], TopicTable = [{Topic, Opts}],
case emqx_mgmt:subscribe(ClientID, TopicTable) of case emqx_mgmt:subscribe(ClientID, TopicTable) of
{error, Reason} -> {error, Reason} ->
{error, Reason}; {error, Reason};

12
apps/emqx_management/test/emqx_mgmt_api_clients_SUITE.erl
View File

@ -43,7 +43,9 @@ t_clients(_) ->
AuthHeader = emqx_mgmt_api_test_util:auth_header_(), AuthHeader = emqx_mgmt_api_test_util:auth_header_(),
{ok, C1} = emqtt:start_link(#{username => Username1, clientid => ClientId1}), {ok, C1} = emqtt:start_link(#{
username => Username1, clientid => ClientId1, proto_ver => v5
}),
{ok, _} = emqtt:connect(C1), {ok, _} = emqtt:connect(C1),
{ok, C2} = emqtt:start_link(#{username => Username2, clientid => ClientId2}), {ok, C2} = emqtt:start_link(#{username => Username2, clientid => ClientId2}),
{ok, _} = emqtt:connect(C2), {ok, _} = emqtt:connect(C2),
@ -87,7 +89,7 @@ t_clients(_) ->
?assertEqual("[]", Client1AuthzCache), ?assertEqual("[]", Client1AuthzCache),
%% post /clients/:clientid/subscribe %% post /clients/:clientid/subscribe
SubscribeBody = #{topic => Topic, qos => Qos}, SubscribeBody = #{topic => Topic, qos => Qos, nl => 1, rh => 1},
SubscribePath = emqx_mgmt_api_test_util:api_path([ SubscribePath = emqx_mgmt_api_test_util:api_path([
"clients", "clients",
binary_to_list(ClientId1), binary_to_list(ClientId1),
@ -121,9 +123,9 @@ t_clients(_) ->
?assertMatch( ?assertMatch(
#{ #{
<<"clientid">> := ClientId1, <<"clientid">> := ClientId1,
<<"nl">> := _, <<"nl">> := 1,
<<"rap">> := _, <<"rap">> := 0,
<<"rh">> := _, <<"rh">> := 1,
<<"node">> := _, <<"node">> := _,
<<"qos">> := Qos, <<"qos">> := Qos,
<<"topic">> := Topic <<"topic">> := Topic