refactor(emqx_topic_metrics): Decorate remote procedure calls

apps/emqx_modules/src/emqx_topic_metrics_api.erl

@@ -183,7 +183,8 @@ operate_topic_metrics(delete, #{bindings := #{topic := Topic0}}) ->
 %%--------------------------------------------------------------------
 
 cluster_accumulation_metrics() ->
-    case multicall(emqx_topic_metrics, metrics, []) of
+    Nodes = mria_mnesia:running_nodes(),
+    case emqx_topic_metrics_proto_v1:metrics(Nodes) of
         {SuccResList, []} ->
             {ok, accumulate_nodes_metrics(SuccResList)};
         {_, FailedNodes} ->
@@ -191,7 +192,8 @@ cluster_accumulation_metrics() ->
     end.
 
 cluster_accumulation_metrics(Topic) ->
-    case multicall(emqx_topic_metrics, metrics, [Topic]) of
+    Nodes = mria_mnesia:running_nodes(),
+    case emqx_topic_metrics_proto_v1:metrics(Nodes, Topic) of
         {SuccResList, []} ->
             case lists:filter(fun({error, _}) -> false; (_) -> true
                               end, SuccResList) of
@@ -244,11 +246,13 @@ do_accumulation_metrics(MetricsIn, {MetricsAcc, _}) ->
     end, #{}, Keys).
 
 reset() ->
-    _ = multicall(emqx_topic_metrics, reset, []),
+    Nodes = mria_mnesia:running_nodes(),
+    _ = emqx_topic_metrics_proto_v1:reset(Nodes),
     ok.
 
 reset(Topic) ->
-    case multicall(emqx_topic_metrics, reset, [Topic]) of
+    Nodes = mria_mnesia:running_nodes(),
+    case emqx_topic_metrics_proto_v1:reset(Nodes, Topic) of
         {SuccResList, []} ->
             case lists:filter(fun({error, _}) -> true; (_) -> false
                               end, SuccResList) of
@@ -262,9 +266,6 @@ reset(Topic) ->
 %%--------------------------------------------------------------------
 %% utils
 
-multicall(M, F, A) ->
-    emqx_rpc:multicall(mria_mnesia:running_nodes(), M, F, A).
-
 reason2httpresp(quota_exceeded) ->
     Msg = list_to_binary(
             io_lib:format("Max topic metrics count is ~p",

apps/emqx_modules/src/proto/emqx_topic_metrics_proto_v1.erl

@@ -0,0 +1,48 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2022 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(emqx_topic_metrics_proto_v1).
+
+-behaviour(emqx_bpapi).
+
+-export([ introduced_in/0
+
+        , metrics/1
+        , metrics/2
+        , reset/1
+        , reset/2
+        ]).
+
+-include_lib("emqx/include/bpapi.hrl").
+
+introduced_in() ->
+    "5.0.0".
+
+-spec metrics([node()]) -> emqx_rpc:multicall_result().
+metrics(Nodes) ->
+    emqx_rpc:multicall(Nodes, emqx_topic_metrics, metrics, []).
+
+-spec metrics([node()], emqx_types:topic()) -> emqx_rpc:multicall_result().
+metrics(Nodes, Topic) ->
+    emqx_rpc:multicall(Nodes, emqx_topic_metrics, metrics, [Topic]).
+
+-spec reset([node()]) -> emqx_rpc:multicall_result().
+reset(Nodes) ->
+    emqx_rpc:multicall(Nodes, emqx_topic_metrics, reset, []).
+
+-spec reset([node()], emqx_types:topic()) -> emqx_rpc:multicall_result().
+reset(Nodes, Topic) ->
+    emqx_rpc:multicall(Nodes, emqx_topic_metrics, reset, [Topic]).