Update to use abstraction function rabbit_control_main:emitting_map/4.

References #62

7 files changed, 21 insertions, 23 deletions

diff --git a/src/rabbit_binding.erl b/src/rabbit_binding.erl
index 50df60c899..9a62fde757 100644
--- a/src/rabbit_binding.erl
+++ b/src/rabbit_binding.erl
@@ -287,9 +287,8 @@ info_all(VHostPath) -> map(VHostPath, fun (B) -> info(B) end).
 info_all(VHostPath, Items) -> map(VHostPath, fun (B) -> info(B, Items) end).
 
 info_all(VHostPath, Items, Ref, AggregatorPid) ->
-    map(VHostPath, fun (B) -> AggregatorPid ! {Ref, info(B, Items)} end),
-    AggregatorPid ! {Ref, finished},
-    ok.
+    rabbit_control_main:emitting_map(
+      AggregatorPid, Ref, fun(B) -> info(B, Items) end, list(VHostPath)).
 
 has_for_source(SrcName) ->
     Match = #route{binding = #binding{source = SrcName, _ = '_'}},
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index e7947493da..670d7a9838 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -328,10 +328,11 @@ info_all(Items) ->
     rabbit_misc:filter_exit_map(fun (C) -> info(C, Items) end, list()).
 
 info_all(Items, Ref, AggregatorPid) ->
-    rabbit_misc:filter_exit_map(fun (C) -> AggregatorPid !
-                                               {Ref, info(C, Items)} end,
-                                list()),
-    AggregatorPid ! {Ref, finished},
+    rabbit_misc:filter_exit_map(
+      fun(E) -> rabbit_control_main:emitting_map(
+                  AggregatorPid, Ref, fun(C) -> info(C, Items) end, [E])
+      end,
+      list()),
     ok.
 
 refresh_config_local() ->
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 9c034e52e3..e387c30239 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -344,9 +344,8 @@ info_all(VHostPath) -> map(VHostPath, fun (X) -> info(X) end).
 info_all(VHostPath, Items) -> map(VHostPath, fun (X) -> info(X, Items) end).
 
 info_all(VHostPath, Items, Ref, AggregatorPid) ->
-    map(VHostPath, fun (X) -> AggregatorPid ! {Ref, info(X, Items)} end),
-    AggregatorPid ! {Ref, finished},
-    ok.
+    rabbit_control_main:emitting_map(
+      AggregatorPid, Ref, fun(X) -> info(X, Items) end, list(VHostPath)).
 
 route(#exchange{name = #resource{virtual_host = VHost, name = RName} = XName,
                 decorators = Decorators} = X,
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 3abf36b52e..7ccf7aadeb 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -443,8 +443,10 @@ connection_info_all() -> cmap(fun (Q) -> connection_info(Q) end).
 connection_info_all(Items) -> cmap(fun (Q) -> connection_info(Q, Items) end).
 
 connection_info_all(Items, Ref, AggregatorPid) ->
-    cmap(fun (Q) -> AggregatorPid ! {Ref, connection_info(Q, Items)} end),
-    AggregatorPid ! {Ref, finished},
+    cmap(fun(E) -> rabbit_control_main:emitting_map(
+                     AggregatorPid, Ref,
+                     fun(Q) -> connection_info(Q, Items) end, [E])
+         end),
     ok.
 
 close_connection(Pid, Explanation) ->
diff --git a/src/rabbit_policy.erl b/src/rabbit_policy.erl
index 51a1d65390..bb6bfea178 100644
--- a/src/rabbit_policy.erl
+++ b/src/rabbit_policy.erl
@@ -171,10 +171,9 @@ list_formatted(VHost) ->
     order_policies(list0(VHost, fun format/1)).
 
 list_formatted(VHost, Ref, AggregatorPid) ->
-    [AggregatorPid ! {Ref, p(P, fun format/1)} ||
-        P <- rabbit_runtime_parameters:list(VHost, <<"policy">>)],
-    AggregatorPid ! {Ref, finished},
-    ok.
+    rabbit_control_main:emitting_map(
+      AggregatorPid, Ref, fun(P) -> p(P, fun format/1) end,
+      rabbit_runtime_parameters:list(VHost, <<"policy">>)).
 
 list0(VHost, DefnFun) ->
     [p(P, DefnFun) || P <- rabbit_runtime_parameters:list(VHost, <<"policy">>)].
diff --git a/src/rabbit_runtime_parameters.erl b/src/rabbit_runtime_parameters.erl
index 15d1dd9cba..2a2bc1e5f3 100644
--- a/src/rabbit_runtime_parameters.erl
+++ b/src/rabbit_runtime_parameters.erl
@@ -200,10 +200,9 @@ list_formatted(VHost) ->
     [pset(value, format(pget(value, P)), P) || P <- list(VHost)].
 
 list_formatted(VHost, Ref, AggregatorPid) ->
-    [AggregatorPid !
-         {Ref, pset(value, format(pget(value, P)), P)} || P <- list(VHost)],
-    AggregatorPid ! {Ref, finished},
-    ok.
+    rabbit_control_main:emitting_map(
+      AggregatorPid, Ref,
+      fun(P) -> pset(value, format(pget(value, P)), P) end, list(VHost)).
 
 lookup(VHost, Component, Name) ->
     case lookup0({VHost, Component, Name}, rabbit_misc:const(not_found)) of
diff --git a/src/rabbit_vhost.erl b/src/rabbit_vhost.erl
index 8dd64ce22f..ab0e8c8764 100644
--- a/src/rabbit_vhost.erl
+++ b/src/rabbit_vhost.erl
@@ -158,6 +158,5 @@ info_all(Items) -> [info(VHost, Items) || VHost <- list()].
 
 info_all(Ref, AggregatorPid)        -> info_all(?INFO_KEYS, Ref, AggregatorPid).
 info_all(Items, Ref, AggregatorPid) ->
-    [AggregatorPid ! {Ref, info(VHost, Items)} || VHost <- list()],
-    AggregatorPid ! {Ref, finished},
-    ok.
+    rabbit_control_main:emitting_map(
+       AggregatorPid, Ref, fun(VHost) -> info(VHost, Items) end, list()).