Merge branch 'stable' into rabbitmq-server-178

9 files changed, 292 insertions, 116 deletions

diff --git a/src/file_handle_cache.erl b/src/file_handle_cache.erl
index b323842ba0..8be19e5be3 100644
--- a/src/file_handle_cache.erl
+++ b/src/file_handle_cache.erl
@@ -148,7 +148,7 @@
          copy/3, set_maximum_since_use/1, delete/1, clear/1]).
 -export([obtain/0, obtain/1, release/0, release/1, transfer/1, transfer/2,
          set_limit/1, get_limit/0, info_keys/0, with_handle/1, with_handle/2,
-         info/0, info/1]).
+         info/0, info/1, clear_read_cache/0]).
 -export([ulimit/0]).
 
 -export([start_link/0, start_link/2, init/1, handle_call/3, handle_cast/2,
@@ -164,6 +164,8 @@
 -define(CLIENT_ETS_TABLE, file_handle_cache_client).
 -define(ELDERS_ETS_TABLE, file_handle_cache_elders).
 
+-include("rabbit.hrl"). % For #amqqueue record definition.
+
 %%----------------------------------------------------------------------------
 
 -record(file,
@@ -581,6 +583,42 @@ info_keys() -> ?INFO_KEYS.
 info() -> info(?INFO_KEYS).
 info(Items) -> gen_server2:call(?SERVER, {info, Items}, infinity).
 
+clear_read_cache() ->
+    gen_server2:cast(?SERVER, clear_read_cache),
+    clear_vhost_read_cache(rabbit_vhost:list()).
+
+clear_vhost_read_cache([]) ->
+    ok;
+clear_vhost_read_cache([VHost | Rest]) ->
+    clear_queue_read_cache(rabbit_amqqueue:list(VHost)),
+    clear_vhost_read_cache(Rest).
+
+clear_queue_read_cache([]) ->
+    ok;
+clear_queue_read_cache([#amqqueue{pid = MPid, slave_pids = SPids} | Rest]) ->
+    %% Limit the action to the current node.
+    Pids = [P || P <- [MPid | SPids], node(P) =:= node()],
+    %% This function is executed in the context of the backing queue
+    %% process because the read buffer is stored in the process
+    %% dictionary.
+    Fun = fun(_, State) ->
+                  clear_process_read_cache(),
+                  State
+          end,
+    [rabbit_amqqueue:run_backing_queue(Pid, rabbit_variable_queue, Fun)
+     || Pid <- Pids],
+    clear_queue_read_cache(Rest).
+
+clear_process_read_cache() ->
+    [
+     begin
+         Handle1 = reset_read_buffer(Handle),
+         put({Ref, fhc_handle}, Handle1)
+     end ||
+        {{Ref, fhc_handle}, Handle} <- get(),
+        size(Handle#handle.read_buffer) > 0
+    ].
+
 %%----------------------------------------------------------------------------
 %% Internal functions
 %%----------------------------------------------------------------------------
@@ -1147,7 +1185,11 @@ handle_cast({transfer, N, FromPid, ToPid}, State) ->
     {noreply, process_pending(
                 update_counts({obtain, socket}, ToPid, +N,
                               update_counts({obtain, socket}, FromPid, -N,
-                                            State)))}.
+                                            State)))};
+
+handle_cast(clear_read_cache, State) ->
+    clear_process_read_cache(),
+    {noreply, State}.
 
 handle_info(check_counts, State) ->
     {noreply, maybe_reduce(State #fhc_state { timer_ref = undefined })};
diff --git a/src/gen_server2.erl b/src/gen_server2.erl
index d2f96b5250..fd0e6553b5 100644
--- a/src/gen_server2.erl
+++ b/src/gen_server2.erl
@@ -576,10 +576,11 @@ init_it(Starter, Parent, Name0, Mod, Args, Options) ->
         {ok, State, Timeout, Backoff = {backoff, _, _, _}, Mod1} ->
             Backoff1 = extend_backoff(Backoff),
             proc_lib:init_ack(Starter, {ok, self()}),
-            loop(GS2State #gs2_state { mod           = Mod1,
-                                       state         = State,
-                                       time          = Timeout,
-                                       timeout_state = Backoff1 });
+            loop(find_prioritisers(
+                  GS2State #gs2_state { mod           = Mod1,
+                                        state         = State,
+                                        time          = Timeout,
+                                        timeout_state = Backoff1 }));
         {stop, Reason} ->
             %% For consistency, we must make sure that the
             %% registered name (if any) is unregistered before
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 9ce800023f..5bfa006e09 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -782,15 +782,41 @@ on_node_up(Node) ->
            fun () ->
                    Qs = mnesia:match_object(rabbit_queue,
                                             #amqqueue{_ = '_'}, write),
-                   [case lists:member(Node, RSs) of
-                        true  -> RSs1 = RSs -- [Node],
-                                 store_queue(
-                                   Q#amqqueue{recoverable_slaves = RSs1});
-                        false -> ok
-                    end || #amqqueue{recoverable_slaves = RSs} = Q <- Qs],
+                   [maybe_clear_recoverable_node(Node, Q) || Q <- Qs],
                    ok
            end).
 
+maybe_clear_recoverable_node(Node,
+                             #amqqueue{sync_slave_pids    = SPids,
+                                       recoverable_slaves = RSs} = Q) ->
+    case lists:member(Node, RSs) of
+        true  ->
+            %% There is a race with
+            %% rabbit_mirror_queue_slave:record_synchronised/1 called
+            %% by the incoming slave node and this function, called
+            %% by the master node. If this function is executed after
+            %% record_synchronised/1, the node is erroneously removed
+            %% from the recoverable slaves list.
+            %%
+            %% We check if the slave node's queue PID is alive. If it is
+            %% the case, then this function is executed after. In this
+            %% situation, we don't touch the queue record, it is already
+            %% correct.
+            DoClearNode =
+                case [SP || SP <- SPids, node(SP) =:= Node] of
+                    [SPid] -> not rabbit_misc:is_process_alive(SPid);
+                    _      -> true
+                end,
+            if
+                DoClearNode -> RSs1 = RSs -- [Node],
+                               store_queue(
+                                 Q#amqqueue{recoverable_slaves = RSs1});
+                true        -> ok
+            end;
+        false ->
+            ok
+    end.
+
 on_node_down(Node) ->
     rabbit_misc:execute_mnesia_tx_with_tail(
       fun () -> QsDels =
diff --git a/src/rabbit_backing_queue.erl b/src/rabbit_backing_queue.erl
index f549ee2d59..d6cd3ca43d 100644
--- a/src/rabbit_backing_queue.erl
+++ b/src/rabbit_backing_queue.erl
@@ -252,10 +252,11 @@
 
 behaviour_info(callbacks) ->
     [{start, 1}, {stop, 0}, {init, 3}, {terminate, 2},
-     {delete_and_terminate, 2}, {purge, 1}, {purge_acks, 1}, {publish, 6},
+     {delete_and_terminate, 2}, {delete_crashed, 1}, {purge, 1},
+     {purge_acks, 1}, {publish, 6},
      {publish_delivered, 5}, {discard, 4}, {drain_confirmed, 1},
-     {dropwhile, 2}, {fetchwhile, 4},
-     {fetch, 2}, {ack, 2}, {requeue, 2}, {ackfold, 4}, {fold, 3}, {len, 1},
+     {dropwhile, 2}, {fetchwhile, 4}, {fetch, 2},
+     {drop, 2}, {ack, 2}, {requeue, 2}, {ackfold, 4}, {fold, 3}, {len, 1},
      {is_empty, 1}, {depth, 1}, {set_ram_duration_target, 2},
      {ram_duration, 1}, {needs_timeout, 1}, {timeout, 1},
      {handle_pre_hibernate, 1}, {resume, 1}, {msg_rates, 1},
diff --git a/src/rabbit_cli.erl b/src/rabbit_cli.erl
index 58724af850..33098ce16b 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]).
+         parse_arguments/4, rpc_call/4, rpc_call/5]).
 
 %%----------------------------------------------------------------------------
 
@@ -94,8 +94,13 @@ main(ParseFun, DoFun, UsageMod) ->
             print_error("~p", [Reason]),
             rabbit_misc:quit(2);
         {badrpc, Reason} ->
-            print_error("unable to connect to node ~w: ~w", [Node, Reason]),
-            print_badrpc_diagnostics([Node]),
+            case Reason of
+                timeout ->
+                    print_error("operation ~w on node ~w timed out", [Command, Node]);
+                _ ->
+                    print_error("unable to connect to node ~w: ~w", [Node, Reason]),
+                    print_badrpc_diagnostics([Node])
+            end,
             rabbit_misc:quit(2);
         {badrpc_multi, Reason, Nodes} ->
             print_error("unable to connect to nodes ~p: ~w", [Nodes, Reason]),
@@ -210,8 +215,11 @@ print_badrpc_diagnostics(Nodes) ->
 %% a timeout unless we set our ticktime to be the same. So let's do
 %% that.
 rpc_call(Node, Mod, Fun, Args) ->
-    case rpc:call(Node, net_kernel, get_net_ticktime, [], ?RPC_TIMEOUT) of
+    rpc_call(Node, Mod, Fun, Args, ?RPC_TIMEOUT).
+
+rpc_call(Node, Mod, Fun, Args, Timeout) ->
+    case rpc:call(Node, net_kernel, get_net_ticktime, [], Timeout) of
         {badrpc, _} = E -> E;
         Time            -> net_kernel:set_net_ticktime(Time, 0),
-                           rpc:call(Node, Mod, Fun, Args, ?RPC_TIMEOUT)
+                           rpc:call(Node, Mod, Fun, Args, Timeout)
     end.
diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
index 25a0fbf9a6..71d0e03130 100644
--- a/src/rabbit_control_main.erl
+++ b/src/rabbit_control_main.erl
@@ -21,11 +21,11 @@
 -export([start/0, stop/0, parse_arguments/2, action/5,
          sync_queue/1, cancel_sync_queue/1, become/1]).
 
--import(rabbit_cli, [rpc_call/4]).
+-import(rabbit_cli, [rpc_call/4, rpc_call/5]).
 
 -define(EXTERNAL_CHECK_INTERVAL, 1000).
 
--define(GLOBAL_DEFS(Node), [?QUIET_DEF, ?NODE_DEF(Node)]).
+-define(GLOBAL_DEFS(Node), [?QUIET_DEF, ?NODE_DEF(Node), ?TIMEOUT_DEF]).
 
 -define(COMMANDS,
         [stop,
@@ -108,6 +108,11 @@
          forget_cluster_node, rename_cluster_node, cluster_status, status,
          environment, eval, force_boot]).
 
+-define(COMMANDS_WITH_TIMEOUT,
+        [list_user_permissions, list_policies, list_queues, list_exchanges,
+        list_bindings, list_connections, list_channels, list_consumers,
+        list_vhosts, list_parameters]).
+
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
@@ -119,6 +124,11 @@
          fun ((string(), [any()]) -> 'ok'))
         -> 'ok').
 
+-spec(action/6 ::
+        (atom(), node(), [string()], [{string(), any()}],
+         fun ((string(), [any()]) -> 'ok'), timeout())
+        -> 'ok').
+
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -136,7 +146,19 @@ start() ->
                                             io:format(Format ++ " ...~n", Args1)
                                     end
                        end,
-              do_action(Command, Node, Args, Opts, Inform)
+              try
+                  T = case get_timeout(Opts) of
+                          {ok, Timeout} ->
+                              Timeout;
+                          {error, _} ->
+                              %% since this is an error with user input, ignore the quiet
+                              %% setting
+                              io:format("Failed to parse provided timeout value, using ~s~n", [?RPC_TIMEOUT]),
+                              ?RPC_TIMEOUT
+                  end,
+                  do_action(Command, Node, Args, Opts, Inform, T)
+              catch _:E -> E
+              end
       end, rabbit_ctl_usage).
 
 parse_arguments(CmdLine, NodeStr) ->
@@ -160,18 +182,64 @@ print_report0(Node, {Module, InfoFun, KeysFun}, VHostArg) ->
     end,
     io:nl().
 
+get_timeout(Opts) ->
+    parse_timeout(proplists:get_value(?TIMEOUT_OPT, Opts, ?RPC_TIMEOUT)).
+
+parse_number(N) when is_list(N) ->
+    try list_to_integer(N) of
+        Val -> Val
+    catch error:badarg ->
+            %% could have been a float, give it
+            %% another shot
+            list_to_float(N)
+    end.
+
+parse_timeout("infinity") ->
+    {ok, infinity};
+parse_timeout(infinity) ->
+    {ok, infinity};
+parse_timeout(N) when is_list(N) ->
+    try parse_number(N) of
+        M ->
+            Y = case M >= 0 of
+                    true  -> round(M) * 1000;
+                    false -> ?RPC_TIMEOUT
+                end,
+            {ok, Y}
+    catch error:badarg ->
+        {error, infinity}
+    end;
+parse_timeout(N) ->
+    {ok, N}.
+
+announce_timeout(infinity, _Inform) ->
+    %% no-op
+    ok;
+announce_timeout(Timeout, Inform) when is_number(Timeout) ->
+    Inform("Timeout: ~w seconds", [Timeout/1000]),
+    ok.
+
 stop() ->
     ok.
 
 %%----------------------------------------------------------------------------
 
-do_action(Command, Node, Args, Opts, Inform) ->
+do_action(Command, Node, Args, Opts, Inform, Timeout) ->
     case lists:member(Command, ?COMMANDS_NOT_REQUIRING_APP) of
-        false -> case ensure_app_running(Node) of
-                     ok -> action(Command, Node, Args, Opts, Inform);
-                     E  -> E
-                 end;
-        true  -> action(Command, Node, Args, Opts, Inform)
+        false ->
+            case ensure_app_running(Node) of
+                ok ->
+                    case lists:member(Command, ?COMMANDS_WITH_TIMEOUT) of
+                        true  ->
+                            announce_timeout(Timeout, Inform),
+                            action(Command, Node, Args, Opts, Inform, Timeout);
+                        false ->
+                            action(Command, Node, Args, Opts, Inform)
+                    end;
+                E  -> E
+            end;
+        true  ->
+            action(Command, Node, Args, Opts, Inform)
     end.
 
 action(stop, Node, Args, _Opts, Inform) ->
@@ -313,12 +381,6 @@ action(set_user_tags, Node, [Username | TagsStr], _Opts, Inform) ->
     rpc_call(Node, rabbit_auth_backend_internal, set_tags,
              [list_to_binary(Username), Tags]);
 
-action(list_users, Node, [], _Opts, Inform) ->
-    Inform("Listing users", []),
-    display_info_list(
-      call(Node, {rabbit_auth_backend_internal, list_users, []}),
-      rabbit_auth_backend_internal:user_info_keys());
-
 action(add_vhost, Node, Args = [_VHostPath], _Opts, Inform) ->
     Inform("Creating vhost \"~s\"", Args),
     call(Node, {rabbit_vhost, add, Args});
@@ -327,63 +389,6 @@ action(delete_vhost, Node, Args = [_VHostPath], _Opts, Inform) ->
     Inform("Deleting vhost \"~s\"", Args),
     call(Node, {rabbit_vhost, delete, Args});
 
-action(list_vhosts, Node, Args, _Opts, Inform) ->
-    Inform("Listing vhosts", []),
-    ArgAtoms = default_if_empty(Args, [name]),
-    display_info_list(call(Node, {rabbit_vhost, info_all, []}), ArgAtoms);
-
-action(list_user_permissions, Node, Args = [_Username], _Opts, Inform) ->
-    Inform("Listing permissions for user ~p", Args),
-    display_info_list(call(Node, {rabbit_auth_backend_internal,
-                                  list_user_permissions, Args}),
-                      rabbit_auth_backend_internal:user_perms_info_keys());
-
-action(list_queues, Node, Args, Opts, Inform) ->
-    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]),
-                      ArgAtoms);
-
-action(list_exchanges, Node, Args, Opts, Inform) ->
-    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]),
-                      ArgAtoms);
-
-action(list_bindings, Node, Args, Opts, Inform) ->
-    Inform("Listing bindings", []),
-    VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
-    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]),
-                      ArgAtoms);
-
-action(list_connections, Node, Args, _Opts, Inform) ->
-    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]),
-                      ArgAtoms);
-
-action(list_channels, Node, Args, _Opts, Inform) ->
-    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]),
-                      ArgAtoms);
-
-action(list_consumers, Node, _Args, Opts, Inform) ->
-    Inform("Listing consumers", []),
-    VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
-    display_info_list(rpc_call(Node, rabbit_amqqueue, consumers_all, [VHostArg]),
-                      rabbit_amqqueue:consumer_info_keys());
-
 action(trace_on, Node, [], Opts, Inform) ->
     VHost = proplists:get_value(?VHOST_OPT, Opts),
     Inform("Starting tracing for vhost \"~s\"", [VHost]),
@@ -416,13 +421,6 @@ action(clear_permissions, Node, [Username], Opts, Inform) ->
     call(Node, {rabbit_auth_backend_internal, clear_permissions,
                 [Username, VHost]});
 
-action(list_permissions, Node, [], Opts, Inform) ->
-    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]}),
-                      rabbit_auth_backend_internal:vhost_perms_info_keys());
-
 action(set_parameter, Node, [Component, Key, Value], Opts, Inform) ->
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
     Inform("Setting runtime parameter ~p for component ~p to ~p",
@@ -438,13 +436,6 @@ action(clear_parameter, Node, [Component, Key], Opts, Inform) ->
                                                       list_to_binary(Component),
                                                       list_to_binary(Key)]);
 
-action(list_parameters, Node, [], Opts, Inform) ->
-    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]),
-      rabbit_runtime_parameters:info_keys());
-
 action(set_policy, Node, [Key, Pattern, Defn], Opts, Inform) ->
     Msg = "Setting policy ~p for pattern ~p to ~p with priority ~p",
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
@@ -460,12 +451,6 @@ action(clear_policy, Node, [Key], Opts, Inform) ->
     Inform("Clearing policy ~p", [Key]),
     rpc_call(Node, rabbit_policy, delete, [VHostArg, list_to_binary(Key)]);
 
-action(list_policies, Node, [], Opts, Inform) ->
-    VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
-    Inform("Listing policies", []),
-    display_info_list(rpc_call(Node, rabbit_policy, list_formatted, [VHostArg]),
-                      rabbit_policy:info_keys());
-
 action(report, Node, _Args, _Opts, Inform) ->
     Inform("Reporting server status on ~p~n~n", [erlang:universaltime()]),
     [begin ok = action(Action, N, [], [], Inform), io:nl() end ||
@@ -493,7 +478,104 @@ action(eval, Node, [Expr], _Opts, _Inform) ->
             end;
         {error, E, _} ->
             {error_string, format_parse_error(E)}
-    end.
+    end;
+
+action(Command, Node, Args, Opts, Inform) ->
+    %% For backward compatibility, run commands accepting a timeout with
+    %% the default timeout.
+    action(Command, Node, Args, Opts, Inform, ?RPC_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());
+
+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());
+
+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());
+
+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());
+
+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);
+
+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());
+
+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);
+
+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);
+
+action(list_bindings, Node, Args, Opts, Inform, Timeout) ->
+    Inform("Listing bindings", []),
+    VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
+    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);
+
+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);
+
+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);
+
+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()).
+
 
 format_parse_error({_Line, Mod, Err}) -> lists:flatten(Mod:format_error(Err)).
 
@@ -650,6 +732,9 @@ 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).
+
 list_to_binary_utf8(L) ->
     B = list_to_binary(L),
     case rabbit_binary_parser:validate_utf8(B) of
diff --git a/src/rabbit_log.erl b/src/rabbit_log.erl
index 56e90d30ae..083204df05 100644
--- a/src/rabbit_log.erl
+++ b/src/rabbit_log.erl
@@ -84,6 +84,7 @@ catlevel(Category) ->
 level(debug)   -> 4;
 level(info)    -> 3;
 level(warning) -> 2;
+level(warn)    -> 2;
 level(error)   -> 1;
 level(none)    -> 0.
 
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index fec8f3077b..6a9eafd9fc 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -196,7 +196,17 @@ socket_error(Reason) when is_atom(Reason) ->
     log(error, "Error on AMQP connection ~p: ~s~n",
         [self(), rabbit_misc:format_inet_error(Reason)]);
 socket_error(Reason) ->
-    log(error, "Error on AMQP connection ~p:~n~p~n", [self(), Reason]).
+    Level =
+        case Reason of
+            {ssl_upgrade_error, closed} ->
+                %% The socket was closed while upgrading to SSL.
+                %% This is presumably a TCP healthcheck, so don't log
+                %% it unless specified otherwise.
+                debug;
+            _ ->
+                error
+        end,
+    log(Level, "Error on AMQP connection ~p:~n~p~n", [self(), Reason]).
 
 inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F).
 
@@ -345,6 +355,8 @@ mainloop(Deb, Buf, BufLen, State = #v1{sock = Sock,
                      State#v1{pending_recv = false});
         closed when State#v1.connection_state =:= closed ->
             ok;
+        closed when CS =:= pre_init andalso Buf =:= [] ->
+            stop(tcp_healthcheck, State);
         closed ->
             stop(closed, State);
         {error, Reason} ->
@@ -359,7 +371,7 @@ mainloop(Deb, Buf, BufLen, State = #v1{sock = Sock,
             end
     end.
 
-stop(closed, #v1{connection_state = pre_init} = State) ->
+stop(tcp_healthcheck, State) ->
     %% The connection was closed before any packet was received. It's
     %% probably a load-balancer healthcheck: don't consider this a
     %% failure.
diff --git a/src/supervisor2.erl b/src/supervisor2.erl
index 57c3bfc113..7b9421eb3e 100644
--- a/src/supervisor2.erl
+++ b/src/supervisor2.erl
@@ -137,7 +137,7 @@
 -record(state, {name,
 		strategy               :: strategy(),
 		children = []          :: [child_rec()],
-		dynamics               :: ?DICT:?DICT() | ?SET:?SET(),
+		dynamics               :: ?DICT:?DICT() | ?SETS:?SET(),
 		intensity              :: non_neg_integer(),
 		period                 :: pos_integer(),
 		restarts = [],