fix: unsubscribe api; query params
This commit is contained in:
parent
ca50dfe686
commit
07a4d54a25
|
@ -87,8 +87,8 @@ node_query(Node, Params, {Tab, QsSchema}, QueryFun) ->
|
|||
{_, Rows} = do_query(Node, Qs, QueryFun, Start, Limit+1),
|
||||
Meta = #{page => Page, limit => Limit},
|
||||
NMeta = case CodCnt =:= 0 of
|
||||
true -> Meta#{count => count(Tab), hasnext => length(Rows) > Limit};
|
||||
_ -> Meta#{count => -1, hasnext => length(Rows) > Limit}
|
||||
true -> Meta#{count => count(Tab)};
|
||||
_ -> Meta#{count => length(Rows)}
|
||||
end,
|
||||
#{meta => NMeta, data => lists:sublist(Rows, Limit)}.
|
||||
|
||||
|
@ -120,8 +120,8 @@ cluster_query(Params, {Tab, QsSchema}, QueryFun) ->
|
|||
Rows = do_cluster_query(Nodes, Qs, QueryFun, Start, Limit+1, []),
|
||||
Meta = #{page => Page, limit => Limit},
|
||||
NMeta = case CodCnt =:= 0 of
|
||||
true -> Meta#{count => count(Tab, Nodes), hasnext => length(Rows) > Limit};
|
||||
_ -> Meta#{count => -1, hasnext => length(Rows) > Limit}
|
||||
true -> Meta#{count => count(Tab, Nodes)};
|
||||
_ -> Meta#{count => length(Rows)}
|
||||
end,
|
||||
#{meta => NMeta, data => lists:sublist(Rows, Limit)}.
|
||||
|
||||
|
|
|
@ -32,6 +32,7 @@
|
|||
, subscriptions/2
|
||||
, authz_cache/2
|
||||
, subscribe/2
|
||||
, unsubscribe/2
|
||||
, subscribe_batch/2]).
|
||||
|
||||
-export([ query/3
|
||||
|
@ -70,7 +71,8 @@ apis() ->
|
|||
, client_api()
|
||||
, clients_authz_cache_api()
|
||||
, clients_subscriptions_api()
|
||||
, subscribe_api()].
|
||||
, subscribe_api()
|
||||
, unsubscribe_api()].
|
||||
|
||||
schemas() ->
|
||||
Client = #{
|
||||
|
@ -211,17 +213,7 @@ schemas() ->
|
|||
}
|
||||
}
|
||||
},
|
||||
Subscription = #{
|
||||
subscription => #{
|
||||
type => object,
|
||||
properties => #{
|
||||
topic => #{
|
||||
type => string},
|
||||
qos => #{
|
||||
type => integer,
|
||||
enum => [0,1,2]}}}
|
||||
},
|
||||
[Client, AuthzCache, Subscription].
|
||||
[Client, AuthzCache].
|
||||
|
||||
clients_api() ->
|
||||
Metadata = #{
|
||||
|
@ -370,6 +362,15 @@ clients_authz_cache_api() ->
|
|||
{"/clients/:clientid/authz_cache", Metadata, authz_cache}.
|
||||
|
||||
clients_subscriptions_api() ->
|
||||
SubscriptionSchema = #{
|
||||
type => object,
|
||||
properties => #{
|
||||
topic => #{
|
||||
type => string},
|
||||
qos => #{
|
||||
type => integer,
|
||||
enum => [0,1,2]}}
|
||||
},
|
||||
Metadata = #{
|
||||
get => #{
|
||||
description => <<"Get client subscriptions">>,
|
||||
|
@ -380,10 +381,33 @@ clients_subscriptions_api() ->
|
|||
required => true
|
||||
}],
|
||||
responses => #{
|
||||
<<"200">> => emqx_mgmt_util:response_array_schema(<<"Get client subscriptions">>, subscription)}}
|
||||
<<"200">> =>
|
||||
emqx_mgmt_util:response_array_schema(<<"Get client subscriptions">>, SubscriptionSchema)}}
|
||||
},
|
||||
{"/clients/:clientid/subscriptions", Metadata, subscriptions}.
|
||||
|
||||
unsubscribe_api() ->
|
||||
Metadata = #{
|
||||
post => #{
|
||||
description => <<"Unsubscribe">>,
|
||||
parameters => [
|
||||
#{
|
||||
name => clientid,
|
||||
in => path,
|
||||
schema => #{type => string},
|
||||
required => true
|
||||
}
|
||||
],
|
||||
'requestBody' => emqx_mgmt_util:request_body_schema(#{
|
||||
type => object,
|
||||
properties => #{
|
||||
topic => #{
|
||||
type => string,
|
||||
description => <<"Topic">>}}}),
|
||||
responses => #{
|
||||
<<"404">> => emqx_mgmt_util:response_error_schema(<<"Client id not found">>),
|
||||
<<"200">> => emqx_mgmt_util:response_schema(<<"Unsubscribe ok">>)}}},
|
||||
{"/clients/:clientid/unsubscribe", Metadata, unsubscribe}.
|
||||
subscribe_api() ->
|
||||
Metadata = #{
|
||||
post => #{
|
||||
|
@ -407,32 +431,14 @@ subscribe_api() ->
|
|||
description => <<"QoS">>}}}),
|
||||
responses => #{
|
||||
<<"404">> => emqx_mgmt_util:response_error_schema(<<"Client id not found">>),
|
||||
<<"200">> => emqx_mgmt_util:response_schema(<<"Subscribe ok">>)}},
|
||||
delete => #{
|
||||
description => <<"Unsubscribe">>,
|
||||
parameters => [
|
||||
#{
|
||||
name => clientid,
|
||||
in => path,
|
||||
schema => #{type => string},
|
||||
required => true
|
||||
},
|
||||
#{
|
||||
name => topic,
|
||||
in => query,
|
||||
schema => #{type => string},
|
||||
required => true
|
||||
}
|
||||
],
|
||||
responses => #{
|
||||
<<"404">> => emqx_mgmt_util:response_error_schema(<<"Client id not found">>),
|
||||
<<"200">> => emqx_mgmt_util:response_schema(<<"Unsubscribe ok">>)}}},
|
||||
<<"200">> => emqx_mgmt_util:response_schema(<<"Subscribe ok">>)}}},
|
||||
{"/clients/:clientid/subscribe", Metadata, subscribe}.
|
||||
|
||||
%%%==============================================================================================
|
||||
%% parameters trans
|
||||
clients(get, _Request) ->
|
||||
list(#{}).
|
||||
clients(get, Request) ->
|
||||
Params = cowboy_req:parse_qs(Request),
|
||||
list(Params).
|
||||
|
||||
client(get, Request) ->
|
||||
ClientID = cowboy_req:binding(clientid, Request),
|
||||
|
@ -456,11 +462,13 @@ subscribe(post, Request) ->
|
|||
TopicInfo = emqx_json:decode(Body, [return_maps]),
|
||||
Topic = maps:get(<<"topic">>, TopicInfo),
|
||||
Qos = maps:get(<<"qos">>, TopicInfo, 0),
|
||||
subscribe(#{clientid => ClientID, topic => Topic, qos => Qos});
|
||||
subscribe(#{clientid => ClientID, topic => Topic, qos => Qos}).
|
||||
|
||||
subscribe(delete, Request) ->
|
||||
unsubscribe(post, Request) ->
|
||||
ClientID = cowboy_req:binding(clientid, Request),
|
||||
#{topic := Topic} = cowboy_req:match_qs([topic], Request),
|
||||
{ok, Body, _} = cowboy_req:read_body(Request),
|
||||
TopicInfo = emqx_json:decode(Body, [return_maps]),
|
||||
Topic = maps:get(<<"topic">>, TopicInfo),
|
||||
unsubscribe(#{clientid => ClientID, topic => Topic}).
|
||||
|
||||
%% TODO: batch
|
||||
|
@ -488,7 +496,7 @@ subscriptions(get, Request) ->
|
|||
%% api apply
|
||||
|
||||
list(Params) ->
|
||||
Response = emqx_mgmt_api:cluster_query(maps:to_list(Params), ?CLIENT_QS_SCHEMA, ?query_fun),
|
||||
Response = emqx_mgmt_api:cluster_query(Params, ?CLIENT_QS_SCHEMA, ?query_fun),
|
||||
{200, Response}.
|
||||
|
||||
lookup(#{clientid := ClientID}) ->
|
||||
|
@ -572,7 +580,7 @@ format_channel_info({_, ClientInfo, ClientStats}) ->
|
|||
ClientInfoMap3 = maps:put(ip_address, IpAddress, ClientInfoMap2),
|
||||
ClientInfoMap = maps:put(connected, Connected, ClientInfoMap3),
|
||||
RemoveList = [
|
||||
auth_result
|
||||
auth_result
|
||||
, peername
|
||||
, sockname
|
||||
, peerhost
|
||||
|
@ -581,6 +589,7 @@ format_channel_info({_, ClientInfo, ClientStats}) ->
|
|||
, conn_props
|
||||
, peercert
|
||||
, sockstate
|
||||
, subscriptions
|
||||
, receive_maximum
|
||||
, protocol
|
||||
, is_superuser
|
||||
|
|
|
@ -94,15 +94,15 @@ t_clients(_) ->
|
|||
|
||||
%% post /clients/:clientid/subscribe
|
||||
SubscribeBody = #{topic => Topic, qos => Qos},
|
||||
SubscribePath = emqx_mgmt_api_test_util:api_path(["clients", binary_to_list(ClientId1), "subscribe"]),
|
||||
SubscribePath = emqx_mgmt_api_test_util:api_path(["clients", binary_to_list(ClientId1), "subscribe"]),
|
||||
{ok, _} = emqx_mgmt_api_test_util:request_api(post, SubscribePath, "", AuthHeader, SubscribeBody),
|
||||
timer:sleep(100),
|
||||
[{{_, AfterSubTopic}, #{qos := AfterSubQos}}] = emqx_mgmt:lookup_subscriptions(ClientId1),
|
||||
?assertEqual(AfterSubTopic, Topic),
|
||||
?assertEqual(AfterSubQos, Qos),
|
||||
|
||||
%% delete /clients/:clientid/subscribe
|
||||
UnSubscribeQuery = "topic=" ++ binary_to_list(Topic),
|
||||
{ok, _} = emqx_mgmt_api_test_util:request_api(delete, SubscribePath, UnSubscribeQuery, AuthHeader),
|
||||
%% post /clients/:clientid/unsubscribe
|
||||
UnSubscribePath = emqx_mgmt_api_test_util:api_path(["clients", binary_to_list(ClientId1), "unsubscribe"]),
|
||||
{ok, _} = emqx_mgmt_api_test_util:request_api(post, UnSubscribePath, "", AuthHeader, SubscribeBody),
|
||||
timer:sleep(100),
|
||||
?assertEqual([], emqx_mgmt:lookup_subscriptions(Client1)).
|
||||
|
|
Loading…
Reference in New Issue