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

refactor(mgmt_api): Use Continuation with ets selecting.

This commit is contained in:
JimMoen 2021-10-08 11:11:05 +08:00
parent 5d1772a6b4
commit 9657d7e628
1 changed files with 27 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
apps/emqx_management/src/emqx_mgmt_api.erl
View File

@ -23,8 +23,7 @@
]). ]).
%% first_next query APIs %% first_next query APIs
-export([ params2qs/2 -export([ node_query/5
, node_query/5
, cluster_query/4 , cluster_query/4
, traverse_table/5 , traverse_table/5
, select_table/5 , select_table/5
@ -130,20 +129,6 @@ node_query(Node, Params, Tab, QsSchema, QueryFun) ->
end, end,
#{meta => NMeta, data => lists:sublist(Rows, Limit)}. #{meta => NMeta, data => lists:sublist(Rows, Limit)}.
%% @private
do_query(Node, Tab, Qs, {M,F}, Start, Limit) when Node =:= node() ->
M:F(Tab, Qs, Start, Limit);
do_query(Node, Tab, Qs, QueryFun, Start, Limit) ->
rpc_call(Node, ?MODULE, do_query,
[Node, Tab, Qs, QueryFun, Start, Limit], 50000).
%% @private
rpc_call(Node, M, F, A, T) ->
case rpc:call(Node, M, F, A, T) of
{badrpc, _} = R -> {error, R};
Res -> Res
end.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
%% Cluster Query %% Cluster Query
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------
@ -176,6 +161,28 @@ do_cluster_query([Node|Nodes], Tab, Qs, QueryFun, Start, Limit, Acc) ->
lists:append(lists:reverse([Rows|Acc])) lists:append(lists:reverse([Rows|Acc]))
end. end.
%%--------------------------------------------------------------------
%% Do Query (or rpc query)
%%--------------------------------------------------------------------
%% @private
do_query(Node, Tab, Qs, {M,F}, Continuation, Limit) when Node =:= node() ->
M:F(Tab, Qs, Continuation, Limit);
do_query(Node, Tab, Qs, QueryFun, Continuation, Limit) ->
rpc_call(Node, ?MODULE, do_query,
[Node, Tab, Qs, QueryFun, Continuation, Limit], 50000).
%% @private
rpc_call(Node, M, F, A, T) ->
case rpc:call(Node, M, F, A, T) of
{badrpc, _} = R -> {error, R};
Res -> Res
end.
%%--------------------------------------------------------------------
%% Table Select
%%--------------------------------------------------------------------
traverse_table(Tab, MatchFun, Start, Limit, FmtFun) -> traverse_table(Tab, MatchFun, Start, Limit, FmtFun) ->
ets:safe_fixtable(Tab, true), ets:safe_fixtable(Tab, true),
{NStart, Rows} = traverse_n_by_one(Tab, ets:first(Tab), MatchFun, Start, Limit, []), {NStart, Rows} = traverse_n_by_one(Tab, ets:first(Tab), MatchFun, Start, Limit, []),
@ -238,15 +245,16 @@ select_n_by_one({Rows0, Cons}, Start, Limit, Acc) ->
select_n_by_one(ets:select(Cons), 0, NLimit, [Got|Acc]) select_n_by_one(ets:select(Cons), 0, NLimit, [Got|Acc])
end. end.
%%--------------------------------------------------------------------
%% Internal funcs
%%--------------------------------------------------------------------
params2qs(Params, QsSchema) when is_map(Params) -> params2qs(Params, QsSchema) when is_map(Params) ->
params2qs(maps:to_list(Params), QsSchema); params2qs(maps:to_list(Params), QsSchema);
params2qs(Params, QsSchema) -> params2qs(Params, QsSchema) ->
{Qs, Fuzzy} = pick_params_to_qs(Params, QsSchema, [], []), {Qs, Fuzzy} = pick_params_to_qs(Params, QsSchema, [], []),
{length(Qs) + length(Fuzzy), {Qs, Fuzzy}}. {length(Qs) + length(Fuzzy), {Qs, Fuzzy}}.
%%--------------------------------------------------------------------
%% Internal funcs
pick_params_to_qs([], _, Acc1, Acc2) -> pick_params_to_qs([], _, Acc1, Acc2) ->
NAcc2 = [E || E <- Acc2, not lists:keymember(element(1, E), 1, Acc1)], NAcc2 = [E || E <- Acc2, not lists:keymember(element(1, E), 1, Acc1)],
{lists:reverse(Acc1), lists:reverse(NAcc2)}; {lists:reverse(Acc1), lists:reverse(NAcc2)};