Merge pull request #408 from rabbitmq/rabbitmq-server-62

RabbitMQ server 62

8 files changed, 147 insertions, 65 deletions

diff --git a/src/rabbit_auth_backend_internal.erl b/src/rabbit_auth_backend_internal.erl
index 6babe135fa..ce5474dc9e 100644
--- a/src/rabbit_auth_backend_internal.erl
+++ b/src/rabbit_auth_backend_internal.erl
@@ -30,8 +30,9 @@
 -export([user_info_keys/0, perms_info_keys/0,
          user_perms_info_keys/0, vhost_perms_info_keys/0,
          user_vhost_perms_info_keys/0,
-         list_users/0, list_permissions/0,
-         list_user_permissions/1, list_vhost_permissions/1,
+         list_users/0, list_users/2, list_permissions/0,
+         list_user_permissions/1, list_user_permissions/3,
+         list_vhost_permissions/1, list_vhost_permissions/3,
          list_user_vhost_permissions/2]).
 
 %% for testing
@@ -66,11 +67,16 @@
 -spec(vhost_perms_info_keys/0 :: () -> rabbit_types:info_keys()).
 -spec(user_vhost_perms_info_keys/0 :: () -> rabbit_types:info_keys()).
 -spec(list_users/0 :: () -> [rabbit_types:infos()]).
+-spec(list_users/2 :: (reference(), pid()) -> 'ok').
 -spec(list_permissions/0 :: () -> [rabbit_types:infos()]).
 -spec(list_user_permissions/1 ::
         (rabbit_types:username()) -> [rabbit_types:infos()]).
+-spec(list_user_permissions/3 ::
+        (rabbit_types:username(), reference(), pid()) -> 'ok').
 -spec(list_vhost_permissions/1 ::
         (rabbit_types:vhost()) -> [rabbit_types:infos()]).
+-spec(list_vhost_permissions/3 ::
+        (rabbit_types:vhost(), reference(), pid()) -> 'ok').
 -spec(list_user_vhost_permissions/2 ::
         (rabbit_types:username(), rabbit_types:vhost())
         -> [rabbit_types:infos()]).
@@ -305,26 +311,28 @@ user_perms_info_keys()       -> [vhost | ?PERMS_INFO_KEYS].
 user_vhost_perms_info_keys() -> ?PERMS_INFO_KEYS.
 
 list_users() ->
-    [[{user, Username}, {tags, Tags}] ||
-        #internal_user{username = Username, tags = Tags} <-
-            mnesia:dirty_match_object(rabbit_user, #internal_user{_ = '_'})].
+    [extract_internal_user_params(U) ||
+        U <- mnesia:dirty_match_object(rabbit_user, #internal_user{_ = '_'})].
+
+list_users(Ref, AggregatorPid) ->
+    rabbit_control_misc:emitting_map(
+      AggregatorPid, Ref,
+      fun(U) -> extract_internal_user_params(U) end,
+      mnesia:dirty_match_object(rabbit_user, #internal_user{_ = '_'})).
 
 list_permissions() ->
     list_permissions(perms_info_keys(), match_user_vhost('_', '_')).
 
 list_permissions(Keys, QueryThunk) ->
-    [filter_props(Keys, [{user,      Username},
-                         {vhost,     VHostPath},
-                         {configure, ConfigurePerm},
-                         {write,     WritePerm},
-                         {read,      ReadPerm}]) ||
-        #user_permission{user_vhost = #user_vhost{username     = Username,
-                                                  virtual_host = VHostPath},
-                         permission = #permission{ configure = ConfigurePerm,
-                                                   write     = WritePerm,
-                                                   read      = ReadPerm}} <-
-            %% TODO: use dirty ops instead
-            rabbit_misc:execute_mnesia_transaction(QueryThunk)].
+    [extract_user_permission_params(Keys, U) ||
+        %% TODO: use dirty ops instead
+        U <- rabbit_misc:execute_mnesia_transaction(QueryThunk)].
+
+list_permissions(Keys, QueryThunk, Ref, AggregatorPid) ->
+    rabbit_control_misc:emitting_map(
+      AggregatorPid, Ref, fun(U) -> extract_user_permission_params(Keys, U) end,
+      %% TODO: use dirty ops instead
+      rabbit_misc:execute_mnesia_transaction(QueryThunk)).
 
 filter_props(Keys, Props) -> [T || T = {K, _} <- Props, lists:member(K, Keys)].
 
@@ -333,17 +341,46 @@ list_user_permissions(Username) ->
       user_perms_info_keys(),
       rabbit_misc:with_user(Username, match_user_vhost(Username, '_'))).
 
+list_user_permissions(Username, Ref, AggregatorPid) ->
+    list_permissions(
+      user_perms_info_keys(),
+      rabbit_misc:with_user(Username, match_user_vhost(Username, '_')),
+      Ref, AggregatorPid).
+
 list_vhost_permissions(VHostPath) ->
     list_permissions(
       vhost_perms_info_keys(),
       rabbit_vhost:with(VHostPath, match_user_vhost('_', VHostPath))).
 
+list_vhost_permissions(VHostPath, Ref, AggregatorPid) ->
+    list_permissions(
+      vhost_perms_info_keys(),
+      rabbit_vhost:with(VHostPath, match_user_vhost('_', VHostPath)),
+      Ref, AggregatorPid).
+
 list_user_vhost_permissions(Username, VHostPath) ->
     list_permissions(
       user_vhost_perms_info_keys(),
       rabbit_misc:with_user_and_vhost(
         Username, VHostPath, match_user_vhost(Username, VHostPath))).
 
+extract_user_permission_params(Keys, #user_permission{
+                                        user_vhost =
+                                            #user_vhost{username     = Username,
+                                                        virtual_host = VHostPath},
+                                        permission = #permission{
+                                                        configure = ConfigurePerm,
+                                                        write     = WritePerm,
+                                                        read      = ReadPerm}}) ->
+    filter_props(Keys, [{user,      Username},
+                        {vhost,     VHostPath},
+                        {configure, ConfigurePerm},
+                        {write,     WritePerm},
+                        {read,      ReadPerm}]).
+
+extract_internal_user_params(#internal_user{username = Username, tags = Tags}) ->
+    [{user, Username}, {tags, Tags}].
+
 match_user_vhost(Username, VHostPath) ->
     fun () -> mnesia:match_object(
                 rabbit_user_permission,
diff --git a/src/rabbit_binding.erl b/src/rabbit_binding.erl
index 77a9277c4a..609daf612e 100644
--- a/src/rabbit_binding.erl
+++ b/src/rabbit_binding.erl
@@ -22,7 +22,7 @@
          list_for_source_and_destination/2]).
 -export([new_deletions/0, combine_deletions/2, add_deletion/3,
          process_deletions/1]).
--export([info_keys/0, info/1, info/2, info_all/1, info_all/2]).
+-export([info_keys/0, info/1, info/2, info_all/1, info_all/2, info_all/4]).
 %% these must all be run inside a mnesia tx
 -export([has_for_source/1, remove_for_source/1,
          remove_for_destination/2, remove_transient_for_destination/1]).
@@ -78,6 +78,8 @@
 -spec(info_all/1 :: (rabbit_types:vhost()) -> [rabbit_types:infos()]).
 -spec(info_all/2 ::(rabbit_types:vhost(), rabbit_types:info_keys())
                    -> [rabbit_types:infos()]).
+-spec(info_all/4 ::(rabbit_types:vhost(), rabbit_types:info_keys(),
+                    reference(), pid()) -> 'ok').
 -spec(has_for_source/1 :: (rabbit_types:binding_source()) -> boolean()).
 -spec(remove_for_source/1 :: (rabbit_types:binding_source()) -> bindings()).
 -spec(remove_for_destination/2 ::
@@ -284,6 +286,10 @@ 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) ->
+    rabbit_control_misc:emitting_map(
+      AggregatorPid, Ref, fun(B) -> info(B, Items) end, list(VHostPath)).
+
 has_for_source(SrcName) ->
     Match = #route{binding = #binding{source = SrcName, _ = '_'}},
     %% we need to check for semi-durable routes (which subsumes
diff --git a/src/rabbit_cli.erl b/src/rabbit_cli.erl
index 6dad003bb5..311906afc6 100644
--- a/src/rabbit_cli.erl
+++ b/src/rabbit_cli.erl
@@ -18,7 +18,7 @@
 -include("rabbit_cli.hrl").
 
 -export([main/3, start_distribution/0, start_distribution/1,
-         parse_arguments/4, rpc_call/4, rpc_call/5]).
+         parse_arguments/4, rpc_call/4, rpc_call/5, rpc_call/7]).
 
 %%----------------------------------------------------------------------------
 
@@ -39,6 +39,9 @@
         ([{atom(), [{string(), optdef()}]} | atom()],
          [{string(), optdef()}], string(), [string()]) -> parse_result()).
 -spec(rpc_call/4 :: (node(), atom(), atom(), [any()]) -> any()).
+-spec(rpc_call/5 :: (node(), atom(), atom(), [any()], number()) -> any()).
+-spec(rpc_call/7 :: (node(), atom(), atom(), [any()], reference(), pid(),
+                     number()) -> any()).
 
 -endif.
 
@@ -110,6 +113,10 @@ main(ParseFun, DoFun, UsageMod) ->
             print_error("unable to connect to nodes ~p: ~w", [Nodes, Reason]),
             print_badrpc_diagnostics(Nodes),
             rabbit_misc:quit(?EX_UNAVAILABLE);
+        function_clause ->
+            print_error("operation ~w used with invalid parameter: ~p",
+                        [Command, Args]),
+            usage(UsageMod);
         {refused, Username, _, _} ->
             print_error("failed to authenticate user \"~s\"", [Username]),
             rabbit_misc:quit(?EX_NOUSER);
@@ -233,3 +240,6 @@ rpc_call(Node, Mod, Fun, Args, Timeout) ->
         Time            -> net_kernel:set_net_ticktime(Time, 0),
                            rpc:call(Node, Mod, Fun, Args, Timeout)
     end.
+
+rpc_call(Node, Mod, Fun, Args, Ref, Pid, Timeout) ->
+    rpc_call(Node, Mod, Fun, Args++[Ref, Pid], Timeout).
diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
index 1d98e5a354..2799d510d0 100644
--- a/src/rabbit_control_main.erl
+++ b/src/rabbit_control_main.erl
@@ -18,11 +18,11 @@
 -include("rabbit.hrl").
 -include("rabbit_cli.hrl").
 
--export([start/0, stop/0, parse_arguments/2, action/5,
+-export([start/0, stop/0, parse_arguments/2, action/5, action/6,
          sync_queue/1, cancel_sync_queue/1, become/1,
          purge_queue/1]).
 
--import(rabbit_cli, [rpc_call/4, rpc_call/5]).
+-import(rabbit_cli, [rpc_call/4, rpc_call/5, rpc_call/7]).
 
 -define(EXTERNAL_CHECK_INTERVAL, 1000).
 
@@ -514,62 +514,53 @@ action(purge_queue, Node, [Q], Opts, Inform, Timeout) ->
 
 action(list_users, Node, [], _Opts, Inform, Timeout) ->
     Inform("Listing users", []),
-    display_info_list(
-      call(Node, {rabbit_auth_backend_internal, list_users, []}, Timeout),
-      rabbit_auth_backend_internal:user_info_keys());
+    call(Node, {rabbit_auth_backend_internal, list_users, []},
+         rabbit_auth_backend_internal:user_info_keys(), true, Timeout);
 
 action(list_permissions, Node, [], Opts, Inform, Timeout) ->
     VHost = proplists:get_value(?VHOST_OPT, Opts),
     Inform("Listing permissions in vhost \"~s\"", [VHost]),
-    display_info_list(call(Node, {rabbit_auth_backend_internal,
-                             list_vhost_permissions, [VHost]}, Timeout),
-                      rabbit_auth_backend_internal:vhost_perms_info_keys());
+    call(Node, {rabbit_auth_backend_internal, list_vhost_permissions, [VHost]},
+         rabbit_auth_backend_internal:vhost_perms_info_keys(), true, Timeout);
 
 action(list_parameters, Node, [], Opts, Inform, Timeout) ->
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
     Inform("Listing runtime parameters", []),
-    display_info_list(
-      rpc_call(Node, rabbit_runtime_parameters, list_formatted, [VHostArg],
-               Timeout),
-      rabbit_runtime_parameters:info_keys());
+    call(Node, {rabbit_runtime_parameters, list_formatted, [VHostArg]},
+         rabbit_runtime_parameters:info_keys(), Timeout);
 
 action(list_policies, Node, [], Opts, Inform, Timeout) ->
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
     Inform("Listing policies", []),
-    display_info_list(rpc_call(Node, rabbit_policy, list_formatted, [VHostArg],
-                              Timeout),
-                      rabbit_policy:info_keys());
+    call(Node, {rabbit_policy, list_formatted, [VHostArg]},
+         rabbit_policy:info_keys(), Timeout);
 
 action(list_vhosts, Node, Args, _Opts, Inform, Timeout) ->
     Inform("Listing vhosts", []),
     ArgAtoms = default_if_empty(Args, [name]),
-    display_info_list(call(Node, {rabbit_vhost, info_all, []}, Timeout),
-                      ArgAtoms);
+    call(Node, {rabbit_vhost, info_all, []}, ArgAtoms, true, Timeout);
 
 action(list_user_permissions, _Node, _Args = [], _Opts, _Inform, _Timeout) ->
     {error_string,
      "list_user_permissions expects a username argument, but none provided."};
 action(list_user_permissions, Node, Args = [_Username], _Opts, Inform, Timeout) ->
     Inform("Listing permissions for user ~p", Args),
-    display_info_list(call(Node, {rabbit_auth_backend_internal,
-                                  list_user_permissions, Args}, Timeout),
-                      rabbit_auth_backend_internal:user_perms_info_keys());
+    call(Node, {rabbit_auth_backend_internal, list_user_permissions, Args},
+         rabbit_auth_backend_internal:user_perms_info_keys(), true, Timeout);
 
 action(list_queues, Node, Args, Opts, Inform, Timeout) ->
     Inform("Listing queues", []),
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
     ArgAtoms = default_if_empty(Args, [name, messages]),
-    display_info_list(rpc_call(Node, rabbit_amqqueue, info_all,
-                               [VHostArg, ArgAtoms], Timeout),
-                      ArgAtoms);
+    call(Node, {rabbit_amqqueue, info_all, [VHostArg, ArgAtoms]},
+         ArgAtoms, Timeout);
 
 action(list_exchanges, Node, Args, Opts, Inform, Timeout) ->
     Inform("Listing exchanges", []),
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
     ArgAtoms = default_if_empty(Args, [name, type]),
-    display_info_list(rpc_call(Node, rabbit_exchange, info_all,
-                               [VHostArg, ArgAtoms], Timeout),
-                      ArgAtoms);
+    call(Node, {rabbit_exchange, info_all, [VHostArg, ArgAtoms]},
+         ArgAtoms, Timeout);
 
 action(list_bindings, Node, Args, Opts, Inform, Timeout) ->
     Inform("Listing bindings", []),
@@ -577,32 +568,27 @@ action(list_bindings, Node, Args, Opts, Inform, Timeout) ->
     ArgAtoms = default_if_empty(Args, [source_name, source_kind,
                                        destination_name, destination_kind,
                                        routing_key, arguments]),
-    display_info_list(rpc_call(Node, rabbit_binding, info_all,
-                               [VHostArg, ArgAtoms], Timeout),
-                      ArgAtoms);
+    call(Node, {rabbit_binding, info_all, [VHostArg, ArgAtoms]},
+         ArgAtoms, Timeout);
 
 action(list_connections, Node, Args, _Opts, Inform, Timeout) ->
     Inform("Listing connections", []),
     ArgAtoms = default_if_empty(Args, [user, peer_host, peer_port, state]),
-    display_info_list(rpc_call(Node, rabbit_networking, connection_info_all,
-                               [ArgAtoms], Timeout),
-                      ArgAtoms);
+    call(Node, {rabbit_networking, connection_info_all, [ArgAtoms]},
+         ArgAtoms, Timeout);
 
 action(list_channels, Node, Args, _Opts, Inform, Timeout) ->
     Inform("Listing channels", []),
     ArgAtoms = default_if_empty(Args, [pid, user, consumer_count,
                                        messages_unacknowledged]),
-    display_info_list(rpc_call(Node, rabbit_channel, info_all, [ArgAtoms],
-                               Timeout),
-                      ArgAtoms);
+    call(Node, {rabbit_channel, info_all, [ArgAtoms]},
+         ArgAtoms, Timeout);
 
 action(list_consumers, Node, _Args, Opts, Inform, Timeout) ->
     Inform("Listing consumers", []),
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
-    display_info_list(rpc_call(Node, rabbit_amqqueue, consumers_all, [VHostArg],
-                               Timeout),
-                      rabbit_amqqueue:consumer_info_keys()).
-
+    call(Node, {rabbit_amqqueue, consumers_all, [VHostArg]},
+         rabbit_amqqueue:consumer_info_keys(), Timeout).
 
 format_parse_error({_Line, Mod, Err}) -> lists:flatten(Mod:format_error(Err)).
 
@@ -694,6 +680,15 @@ default_if_empty(List, Default) when is_list(List) ->
        true       -> [list_to_atom(X) || X <- List]
     end.
 
+display_info_message(Result, InfoItemKeys) ->
+    display_row([format_info_item(
+                   case proplists:lookup(X, Result) of
+                       none when is_list(Result), length(Result) > 0 ->
+                           exit({error, {bad_info_key, X}});
+                       none -> Result;
+                       {X, Value} -> Value
+                   end) || X <- InfoItemKeys]).
+
 display_info_list(Results, InfoItemKeys) when is_list(Results) ->
     lists:foreach(
       fun (Result) -> display_row(
@@ -766,8 +761,18 @@ ensure_app_running(Node) ->
 call(Node, {Mod, Fun, Args}) ->
     rpc_call(Node, Mod, Fun, lists:map(fun list_to_binary_utf8/1, Args)).
 
-call(Node, {Mod, Fun, Args}, Timeout) ->
-    rpc_call(Node, Mod, Fun, lists:map(fun list_to_binary_utf8/1, Args), Timeout).
+call(Node, {Mod, Fun, Args}, InfoKeys, Timeout) ->
+    call(Node, {Mod, Fun, Args}, InfoKeys, false, Timeout).
+
+call(Node, {Mod, Fun, Args}, InfoKeys, ToBinUtf8, Timeout) ->
+    Args0 = case ToBinUtf8 of
+                true  -> lists:map(fun list_to_binary_utf8/1, Args);
+                false -> Args
+            end,
+    spawn_link(rabbit_cli, rpc_call, [Node, Mod, Fun, Args0, Ref = make_ref(),
+                                      Pid = self(), Timeout]),
+    rabbit_control_misc:wait_for_info_messages(
+      Pid, Ref, InfoKeys, fun display_info_message/2, Timeout).
 
 list_to_binary_utf8(L) ->
     B = list_to_binary(L),
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 459334455f..ec5065a1d6 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -22,7 +22,7 @@
          assert_equivalence/6, assert_args_equivalence/2, check_type/1,
          lookup/1, lookup_or_die/1, list/0, list/1, lookup_scratch/2,
          update_scratch/3, update_decorators/1, immutable/1,
-         info_keys/0, info/1, info/2, info_all/1, info_all/2,
+         info_keys/0, info/1, info/2, info_all/1, info_all/2, info_all/4,
          route/2, delete/2, validate_binding/2]).
 %% these must be run inside a mnesia tx
 -export([maybe_auto_delete/2, serial/1, peek_serial/1, update/2]).
@@ -82,6 +82,9 @@
 -spec(info_all/1 :: (rabbit_types:vhost()) -> [rabbit_types:infos()]).
 -spec(info_all/2 ::(rabbit_types:vhost(), rabbit_types:info_keys())
                    -> [rabbit_types:infos()]).
+-spec(info_all/4 ::(rabbit_types:vhost(), rabbit_types:info_keys(),
+                    reference(), pid())
+                   -> 'ok').
 -spec(route/2 :: (rabbit_types:exchange(), rabbit_types:delivery())
                  -> [rabbit_amqqueue:name()]).
 -spec(delete/2 ::
@@ -340,6 +343,10 @@ 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) ->
+    rabbit_control_misc: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,
       #delivery{message = #basic_message{routing_keys = RKs}} = Delivery) ->
diff --git a/src/rabbit_policy.erl b/src/rabbit_policy.erl
index 5bf5483272..dd50095517 100644
--- a/src/rabbit_policy.erl
+++ b/src/rabbit_policy.erl
@@ -29,7 +29,7 @@
 -export([name/1, get/2, get_arg/3, set/1]).
 -export([validate/5, notify/4, notify_clear/3]).
 -export([parse_set/6, set/6, delete/2, lookup/2, list/0, list/1,
-         list_formatted/1, info_keys/0]).
+         list_formatted/1, list_formatted/3, info_keys/0]).
 
 -rabbit_boot_step({?MODULE,
                    [{description, "policy parameters"},
@@ -170,6 +170,10 @@ list(VHost) ->
 list_formatted(VHost) ->
     order_policies(list0(VHost, fun format/1)).
 
+list_formatted(VHost, Ref, AggregatorPid) ->
+    rabbit_control_misc:emitting_map(AggregatorPid, Ref,
+                                     fun(P) -> P end, list_formatted(VHost)).
+
 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 fafd598bb7..5e6f12904d 100644
--- a/src/rabbit_runtime_parameters.erl
+++ b/src/rabbit_runtime_parameters.erl
@@ -19,8 +19,8 @@
 -include("rabbit.hrl").
 
 -export([parse_set/5, set/5, set_any/5, clear/3, clear_any/3, list/0, list/1,
-         list_component/1, list/2, list_formatted/1, lookup/3,
-         value/3, value/4, info_keys/0]).
+         list_component/1, list/2, list_formatted/1, list_formatted/3,
+         lookup/3, value/3, value/4, info_keys/0]).
 
 -export([set_global/2, value_global/1, value_global/2]).
 
@@ -48,6 +48,7 @@
 -spec(list/2 :: (rabbit_types:vhost() | '_', binary() | '_')
                 -> [rabbit_types:infos()]).
 -spec(list_formatted/1 :: (rabbit_types:vhost()) -> [rabbit_types:infos()]).
+-spec(list_formatted/3 :: (rabbit_types:vhost(), reference(), pid()) -> 'ok').
 -spec(lookup/3 :: (rabbit_types:vhost(), binary(), binary())
                   -> rabbit_types:infos() | 'not_found').
 -spec(value/3 :: (rabbit_types:vhost(), binary(), binary()) -> term()).
@@ -198,6 +199,11 @@ list(VHost, Component) ->
 list_formatted(VHost) ->
     [pset(value, format(pget(value, P)), P) || P <- list(VHost)].
 
+list_formatted(VHost, Ref, AggregatorPid) ->
+    rabbit_control_misc: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
         not_found -> not_found;
diff --git a/src/rabbit_vhost.erl b/src/rabbit_vhost.erl
index 9b627adf5d..c1394c321c 100644
--- a/src/rabbit_vhost.erl
+++ b/src/rabbit_vhost.erl
@@ -21,7 +21,7 @@
 %%----------------------------------------------------------------------------
 
 -export([add/1, delete/1, exists/1, list/0, with/2, assert/1]).
--export([info/1, info/2, info_all/0, info_all/1]).
+-export([info/1, info/2, info_all/0, info_all/1, info_all/2, info_all/3]).
 
 -ifdef(use_specs).
 
@@ -37,6 +37,8 @@
                 -> rabbit_types:infos()).
 -spec(info_all/0 :: () -> [rabbit_types:infos()]).
 -spec(info_all/1 :: (rabbit_types:info_keys()) -> [rabbit_types:infos()]).
+-spec(info_all/3 :: (rabbit_types:info_keys(), reference(), pid()) ->
+                         'ok').
 
 -endif.
 
@@ -153,3 +155,8 @@ info(VHost, Items) -> infos(Items, VHost).
 
 info_all()      -> info_all(?INFO_KEYS).
 info_all(Items) -> [info(VHost, Items) || VHost <- list()].
+
+info_all(Ref, AggregatorPid)        -> info_all(?INFO_KEYS, Ref, AggregatorPid).
+info_all(Items, Ref, AggregatorPid) ->
+    rabbit_control_misc:emitting_map(
+       AggregatorPid, Ref, fun(VHost) -> info(VHost, Items) end, list()).