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

refactor(mgmt_api): `cluster_query`.

This commit is contained in:
JimMoen 2021-10-08 14:52:32 +08:00
parent 705ed0ce7b
commit e26804b9ac
1 changed files with 14 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -139,28 +139,29 @@ cluster_query(Params, Tab, QsSchema, QueryFun) ->
{CodCnt, Qs} = params2qs(Params, QsSchema), {CodCnt, Qs} = params2qs(Params, QsSchema),
Limit = b2i(limit(Params)), Limit = b2i(limit(Params)),
Page = b2i(page(Params)), Page = b2i(page(Params)),
Start = if Page > 1 -> (Page-1) * Limit; PageStart = page_start(Page, Limit),
true -> 0
end,
Nodes = ekka_mnesia:running_nodes(), Nodes = ekka_mnesia:running_nodes(),
Rows = do_cluster_query(Nodes, Tab, Qs, QueryFun, Start, Limit+1, []), %% Rows so big, fixme.
Rows = do_cluster_query(Nodes, Tab, Qs, QueryFun, _Continuation = ?FRESH_SELECT, Limit, []),
Data = lists:sublist(Rows, PageStart, Limit),
Meta = #{page => Page, limit => Limit}, Meta = #{page => Page, limit => Limit},
NMeta = case CodCnt =:= 0 of NMeta = case CodCnt =:= 0 of
true -> Meta#{count => lists:sum([rpc_call(Node, ets, info, [Tab, size], 5000) || Node <- Nodes])}; true -> Meta#{count => lists:sum([rpc_call(Node, ets, info, [Tab, size], 5000) || Node <- Nodes])};
_ -> Meta#{count => length(Rows)} _ -> Meta#{count => length(Rows)}
end, end,
#{meta => NMeta, data => lists:sublist(Rows, Limit)}. #{meta => NMeta, data => Data}.
%% @private %% @private
do_cluster_query([], _, _, _, _, _, Acc) -> do_cluster_query([], _Tab, _Qs, _QueryFun, _Continuation, _Limit, Acc) ->
lists:append(lists:reverse(Acc)); lists:append(lists:reverse(Acc));
do_cluster_query([Node|Nodes], Tab, Qs, QueryFun, Start, Limit, Acc) -> do_cluster_query([Node | Nodes], Tab, Qs, QueryFun, Continuation, Limit, Acc) ->
{NStart, Rows} = do_query(Node, Tab, Qs, QueryFun, Start, Limit), {Rows, NContinuation} = do_query(Node, Tab, Qs, QueryFun, Continuation, Limit),
case Limit - length(Rows) of NAcc = [Rows | Acc],
Rest when Rest > 0 -> case NContinuation of
do_cluster_query(Nodes, Tab, Qs, QueryFun, NStart, Limit, [Rows|Acc]); ?FRESH_SELECT ->
0 -> do_cluster_query(Nodes, Tab, Qs, QueryFun, NContinuation, Limit, NAcc);
lists:append(lists:reverse([Rows|Acc])) _ ->
do_cluster_query([Node | Nodes], Tab, Qs, QueryFun, NContinuation, Limit, NAcc)
end. end.
%%-------------------------------------------------------------------- %%--------------------------------------------------------------------