Merging default into bug23727

31 files changed, 1130 insertions, 982 deletions

diff --git a/src/delegate.erl b/src/delegate.erl
index ff55a15b40..17046201ad 100644
--- a/src/delegate.erl
+++ b/src/delegate.erl
@@ -18,7 +18,7 @@
 
 -behaviour(gen_server2).
 
--export([start_link/1, invoke_no_result/2, invoke/2, delegate_count/0]).
+-export([start_link/1, invoke_no_result/2, invoke/2]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
@@ -36,8 +36,6 @@
         ([pid()], fun ((pid()) -> A)) -> {[{pid(), A}],
                                           [{pid(), term()}]}).
 
--spec(delegate_count/0 :: () -> non_neg_integer()).
-
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -68,9 +66,9 @@ invoke(Pids, Fun) when is_list(Pids) ->
     {Replies, BadNodes} =
         case orddict:fetch_keys(Grouped) of
             []          -> {[], []};
-            RemoteNodes -> gen_server2:multi_call(RemoteNodes, delegate(),
-                                                  {invoke, Fun, Grouped},
-                                                  infinity)
+            RemoteNodes -> gen_server2:multi_call(
+                             RemoteNodes, delegate(RemoteNodes),
+                             {invoke, Fun, Grouped}, infinity)
         end,
     BadPids = [{Pid, {exit, {nodedown, BadNode}, []}} ||
                   BadNode <- BadNodes,
@@ -92,7 +90,7 @@ invoke_no_result(Pids, Fun) when is_list(Pids) ->
     {LocalPids, Grouped} = group_pids_by_node(Pids),
     case orddict:fetch_keys(Grouped) of
         []          -> ok;
-        RemoteNodes -> gen_server2:abcast(RemoteNodes, delegate(),
+        RemoteNodes -> gen_server2:abcast(RemoteNodes, delegate(RemoteNodes),
                                           {invoke, Fun, Grouped})
     end,
     safe_invoke(LocalPids, Fun), %% must not die
@@ -111,17 +109,14 @@ group_pids_by_node(Pids) ->
                  node(Pid), fun (List) -> [Pid | List] end, [Pid], Remote)}
       end, {[], orddict:new()}, Pids).
 
-delegate_count() ->
-    {ok, Count} = application:get_env(rabbit, delegate_count),
-    Count.
-
 delegate_name(Hash) ->
     list_to_atom("delegate_" ++ integer_to_list(Hash)).
 
-delegate() ->
+delegate(RemoteNodes) ->
     case get(delegate) of
         undefined -> Name = delegate_name(
-                              erlang:phash2(self(), delegate_count())),
+                              erlang:phash2(self(),
+                                            delegate_sup:count(RemoteNodes))),
                      put(delegate, Name),
                      Name;
         Name      -> Name
diff --git a/src/delegate_sup.erl b/src/delegate_sup.erl
index 5274722145..fc693c7d3d 100644
--- a/src/delegate_sup.erl
+++ b/src/delegate_sup.erl
@@ -18,7 +18,7 @@
 
 -behaviour(supervisor).
 
--export([start_link/0]).
+-export([start_link/1, count/1]).
 
 -export([init/1]).
 
@@ -28,20 +28,32 @@
 
 -ifdef(use_specs).
 
--spec(start_link/0 :: () -> {'ok', pid()} | {'error', any()}).
+-spec(start_link/1 :: (integer()) -> {'ok', pid()} | {'error', any()}).
+-spec(count/1 :: ([node()]) -> integer()).
 
 -endif.
 
 %%----------------------------------------------------------------------------
 
-start_link() ->
-    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+start_link(Count) ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, [Count]).
+
+count([]) ->
+    1;
+count([Node | Nodes]) ->
+    try
+        length(supervisor:which_children({?SERVER, Node}))
+    catch exit:{{R, _}, _} when R =:= nodedown; R =:= shutdown ->
+            count(Nodes);
+          exit:{R, _}      when R =:= noproc; R =:= normal; R =:= shutdown;
+                                R =:= nodedown ->
+            count(Nodes)
+    end.
 
 %%----------------------------------------------------------------------------
 
-init(_Args) ->
-    DCount = delegate:delegate_count(),
+init([Count]) ->
     {ok, {{one_for_one, 10, 10},
           [{Num, {delegate, start_link, [Num]},
             transient, 16#ffffffff, worker, [delegate]} ||
-              Num <- lists:seq(0, DCount - 1)]}}.
+              Num <- lists:seq(0, Count - 1)]}}.
diff --git a/src/file_handle_cache.erl b/src/file_handle_cache.erl
index 1e1f37cb3d..f41815d06b 100644
--- a/src/file_handle_cache.erl
+++ b/src/file_handle_cache.erl
@@ -146,7 +146,8 @@
 -export([open/3, close/1, read/2, append/2, sync/1, position/2, truncate/1,
          last_sync_offset/1, current_virtual_offset/1, current_raw_offset/1,
          flush/1, copy/3, set_maximum_since_use/1, delete/1, clear/1]).
--export([obtain/0, transfer/1, set_limit/1, get_limit/0]).
+-export([obtain/0, transfer/1, set_limit/1, get_limit/0, info_keys/0, info/0,
+         info/1]).
 -export([ulimit/0]).
 
 -export([start_link/0, init/1, handle_call/3, handle_cast/2, handle_info/2,
@@ -259,11 +260,17 @@
 -spec(transfer/1 :: (pid()) -> 'ok').
 -spec(set_limit/1 :: (non_neg_integer()) -> 'ok').
 -spec(get_limit/0 :: () -> non_neg_integer()).
+-spec(info_keys/0 :: () -> [atom()]).
+-spec(info/0 :: () -> [{atom(), any()}]).
+-spec(info/1 :: ([atom()]) -> [{atom(), any()}]).
 -spec(ulimit/0 :: () -> 'infinity' | 'unknown' | non_neg_integer()).
 
 -endif.
 
 %%----------------------------------------------------------------------------
+-define(INFO_KEYS, [obtain_count, obtain_limit]).
+
+%%----------------------------------------------------------------------------
 %% Public API
 %%----------------------------------------------------------------------------
 
@@ -494,6 +501,11 @@ set_limit(Limit) ->
 get_limit() ->
     gen_server:call(?SERVER, get_limit, infinity).
 
+info_keys() -> ?INFO_KEYS.
+
+info() -> info(?INFO_KEYS).
+info(Items) -> gen_server:call(?SERVER, {info, Items}, infinity).
+
 %%----------------------------------------------------------------------------
 %% Internal functions
 %%----------------------------------------------------------------------------
@@ -789,6 +801,12 @@ write_buffer(Handle = #handle { hdl = Hdl, offset = Offset,
             {Error, Handle}
     end.
 
+infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
+
+i(obtain_count, #fhc_state{obtain_count = Count}) -> Count;
+i(obtain_limit, #fhc_state{obtain_limit = Limit}) -> Limit;
+i(Item, _) -> throw({bad_argument, Item}).
+
 %%----------------------------------------------------------------------------
 %% gen_server callbacks
 %%----------------------------------------------------------------------------
@@ -871,13 +889,18 @@ handle_call({obtain, Pid}, From, State = #fhc_state { obtain_count   = Count,
         false ->
             {noreply, run_pending_item(Item, State)}
     end;
+
 handle_call({set_limit, Limit}, _From, State) ->
     {reply, ok, maybe_reduce(
                   process_pending(State #fhc_state {
                                     limit        = Limit,
                                     obtain_limit = obtain_limit(Limit) }))};
+
 handle_call(get_limit, _From, State = #fhc_state { limit = Limit }) ->
-    {reply, Limit, State}.
+    {reply, Limit, State};
+
+handle_call({info, Items}, _From, State) ->
+    {reply, infos(Items, State), State}.
 
 handle_cast({register_callback, Pid, MFA},
             State = #fhc_state { clients = Clients }) ->
diff --git a/src/gen_server2.erl b/src/gen_server2.erl
index a637ddddc8..94296f9751 100644
--- a/src/gen_server2.erl
+++ b/src/gen_server2.erl
@@ -58,6 +58,15 @@
 %% hibernate the process immediately, as it would if backoff wasn't
 %% being used. Instead it'll wait for the current timeout as described
 %% above.
+%%
+%% 7) The callback module can return from any of the handle_*
+%% functions, a {become, Module, State} triple, or a {become, Module,
+%% State, Timeout} quadruple. This allows the gen_server to
+%% dynamically change the callback module. The State is the new state
+%% which will be passed into any of the callback functions in the new
+%% module. Note there is no form also encompassing a reply, thus if
+%% you wish to reply in handle_call/3 and change the callback module,
+%% you need to use gen_server2:reply/2 to issue the reply manually.
 
 %% All modifications are (C) 2009-2011 VMware, Inc.
 
@@ -880,6 +889,22 @@ handle_common_reply(Reply, Msg, GS2State = #gs2_state { name  = Name,
             loop(GS2State #gs2_state { state = NState,
                                        time  = Time1,
                                        debug = Debug1 });
+        {become, Mod, NState} ->
+            Debug1 = common_debug(Debug, fun print_event/3, Name,
+                                  {become, Mod, NState}),
+            loop(find_prioritisers(
+                   GS2State #gs2_state { mod   = Mod,
+                                         state = NState,
+                                         time  = infinity,
+                                         debug = Debug1 }));
+        {become, Mod, NState, Time1} ->
+            Debug1 = common_debug(Debug, fun print_event/3, Name,
+                                  {become, Mod, NState}),
+            loop(find_prioritisers(
+                   GS2State #gs2_state { mod   = Mod,
+                                         state = NState,
+                                         time  = Time1,
+                                         debug = Debug1 }));
         _ ->
             handle_common_termination(Reply, Msg, GS2State)
     end.
diff --git a/src/pg_local.erl b/src/pg_local.erl
index fd515747e5..c9c3a3a715 100644
--- a/src/pg_local.erl
+++ b/src/pg_local.erl
@@ -83,7 +83,7 @@ get_members(Name) ->
 
 sync() ->
     ensure_started(),
-    gen_server:call(?MODULE, sync).
+    gen_server:call(?MODULE, sync, infinity).
 
 %%%
 %%% Callback functions from gen_server
diff --git a/src/rabbit.erl b/src/rabbit.erl
index b041a6372c..faf484af9c 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -27,7 +27,7 @@
 
 %%---------------------------------------------------------------------------
 %% Boot steps.
--export([maybe_insert_default_data/0]).
+-export([maybe_insert_default_data/0, boot_delegate/0]).
 
 -rabbit_boot_step({codec_correctness_check,
                    [{description, "codec correctness check"},
@@ -101,8 +101,7 @@
 
 -rabbit_boot_step({delegate_sup,
                    [{description, "cluster delegate"},
-                    {mfa,         {rabbit_sup, start_child,
-                                   [delegate_sup]}},
+                    {mfa,         {rabbit, boot_delegate, []}},
                     {requires,    kernel_ready},
                     {enables,     core_initialized}]}).
 
@@ -145,13 +144,18 @@
                     {requires,    routing_ready},
                     {enables,     networking}]}).
 
+-rabbit_boot_step({direct_client,
+                   [{mfa,        {rabbit_direct, boot, []}},
+                    {requires,   log_relay}]}).
+
 -rabbit_boot_step({networking,
                    [{mfa,         {rabbit_networking, boot, []}},
-                    {requires,    log_relay},
-                    {enables,     networking_listening}]}).
+                    {requires,    log_relay}]}).
 
--rabbit_boot_step({networking_listening,
-                   [{description, "network listeners available"}]}).
+-rabbit_boot_step({notify_cluster,
+                   [{description, "notify cluster nodes"},
+                    {mfa,         {rabbit_node_monitor, notify_cluster, []}},
+                    {requires,    networking}]}).
 
 %%---------------------------------------------------------------------------
 
@@ -179,6 +183,9 @@
                {running_nodes, [node()]}]).
 -spec(log_location/1 :: ('sasl' | 'kernel') -> log_location()).
 
+-spec(maybe_insert_default_data/0 :: () -> 'ok').
+-spec(boot_delegate/0 :: () -> 'ok').
+
 -endif.
 
 %%----------------------------------------------------------------------------
@@ -207,7 +214,8 @@ stop_and_halt() ->
     ok.
 
 status() ->
-    [{running_applications, application:which_applications()}] ++
+    [{pid, list_to_integer(os:getpid())},
+     {running_applications, application:which_applications()}] ++
         rabbit_mnesia:status().
 
 rotate_logs(BinarySuffix) ->
@@ -225,11 +233,11 @@ start(normal, []) ->
     case erts_version_check() of
         ok ->
             {ok, SupPid} = rabbit_sup:start_link(),
+            true = register(rabbit, self()),
 
             print_banner(),
             [ok = run_boot_step(Step) || Step <- boot_steps()],
             io:format("~nbroker running~n"),
-
             {ok, SupPid};
         Error ->
             Error
@@ -448,6 +456,10 @@ ensure_working_log_handler(OldFHandler, NewFHandler, TTYHandler,
                      end
     end.
 
+boot_delegate() ->
+    {ok, Count} = application:get_env(rabbit, delegate_count),
+    rabbit_sup:start_child(delegate_sup, [Count]).
+
 maybe_insert_default_data() ->
     case rabbit_mnesia:is_db_empty() of
         true -> insert_default_data();
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index ad9e3ce609..46b78c395c 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -137,7 +137,9 @@
         -> queue_or_not_found() | rabbit_misc:thunk(queue_or_not_found())).
 -spec(internal_delete/1 ::
         (name()) -> rabbit_types:ok_or_error('not_found') |
-                    rabbit_types:connection_exit()).
+                    rabbit_types:connection_exit() |
+                    fun ((boolean()) -> rabbit_types:ok_or_error('not_found') |
+                                        rabbit_types:connection_exit())).
 -spec(maybe_run_queue_via_backing_queue/2 ::
         (pid(), (fun ((A) -> {[rabbit_guid:guid()], A}))) -> 'ok').
 -spec(maybe_run_queue_via_backing_queue_async/2 ::
@@ -195,7 +197,7 @@ declare(QueueName, Durable, AutoDelete, Args, Owner) ->
                                       arguments = Args,
                                       exclusive_owner = Owner,
                                       pid = none}),
-    case gen_server2:call(Q#amqqueue.pid, {init, false}) of
+    case gen_server2:call(Q#amqqueue.pid, {init, false}, infinity) of
         not_found -> rabbit_misc:not_found(QueueName);
         Q1        -> Q1
     end.
@@ -215,8 +217,12 @@ internal_declare(Q = #amqqueue{name = QueueName}, false) ->
                           [_] -> %% Q exists on stopped node
                                  rabbit_misc:const(not_found)
                       end;
-                  [ExistingQ] ->
-                      rabbit_misc:const(ExistingQ)
+                  [ExistingQ = #amqqueue{pid = QPid}] ->
+                      case rabbit_misc:is_process_alive(QPid) of
+                          true  -> rabbit_misc:const(ExistingQ);
+                          false -> TailFun = internal_delete(QueueName),
+                                   fun (Tx) -> TailFun(Tx), ExistingQ end
+                      end
               end
       end).
 
@@ -294,29 +300,19 @@ check_declare_arguments(QueueName, Args) ->
                              "invalid arg '~s' for ~s: ~w",
                              [Key, rabbit_misc:rs(QueueName), Error])
      end || {Key, Fun} <-
-                [{<<"x-expires">>,     fun check_expires_argument/1},
-                 {<<"x-message-ttl">>, fun check_message_ttl_argument/1}]],
+                [{<<"x-expires">>,     fun check_integer_argument/1},
+                 {<<"x-message-ttl">>, fun check_integer_argument/1}]],
     ok.
 
-check_expires_argument(Val) ->
-    check_integer_argument(Val,
-                           expires_not_of_acceptable_type,
-                           expires_zero_or_less).
-
-check_message_ttl_argument(Val) ->
-    check_integer_argument(Val,
-                           ttl_not_of_acceptable_type,
-                           ttl_zero_or_less).
-
-check_integer_argument(undefined, _, _) ->
+check_integer_argument(undefined) ->
     ok;
-check_integer_argument({Type, Val}, InvalidTypeError, _) when Val > 0 ->
+check_integer_argument({Type, Val}) when Val > 0 ->
     case lists:member(Type, ?INTEGER_ARG_TYPES) of
         true  -> ok;
-        false -> {error, {InvalidTypeError, Type, Val}}
+        false -> {error, {unacceptable_type, Type}}
     end;
-check_integer_argument({_Type, _Val}, _, ZeroOrLessError) ->
-    {error, ZeroOrLessError}.
+check_integer_argument({_Type, Val}) ->
+    {error, {value_zero_or_less, Val}}.
 
 list(VHostPath) ->
     mnesia:dirty_match_object(
@@ -328,10 +324,10 @@ info_keys() -> rabbit_amqqueue_process:info_keys().
 map(VHostPath, F) -> rabbit_misc:filter_exit_map(F, list(VHostPath)).
 
 info(#amqqueue{ pid = QPid }) ->
-    delegate_call(QPid, info, infinity).
+    delegate_call(QPid, info).
 
 info(#amqqueue{ pid = QPid }, Items) ->
-    case delegate_call(QPid, {info, Items}, infinity) of
+    case delegate_call(QPid, {info, Items}) of
         {ok, Res}      -> Res;
         {error, Error} -> throw(Error)
     end.
@@ -341,7 +337,7 @@ info_all(VHostPath) -> map(VHostPath, fun (Q) -> info(Q) end).
 info_all(VHostPath, Items) -> map(VHostPath, fun (Q) -> info(Q, Items) end).
 
 consumers(#amqqueue{ pid = QPid }) ->
-    delegate_call(QPid, consumers, infinity).
+    delegate_call(QPid, consumers).
 
 consumers_all(VHostPath) ->
     lists:append(
@@ -350,7 +346,8 @@ consumers_all(VHostPath) ->
                          {ChPid, ConsumerTag, AckRequired} <- consumers(Q)]
           end)).
 
-stat(#amqqueue{pid = QPid}) -> delegate_call(QPid, stat, infinity).
+stat(#amqqueue{pid = QPid}) ->
+    delegate_call(QPid, stat).
 
 emit_stats(#amqqueue{pid = QPid}) ->
     delegate_cast(QPid, emit_stats).
@@ -359,9 +356,9 @@ delete_immediately(#amqqueue{ pid = QPid }) ->
     gen_server2:cast(QPid, delete_immediately).
 
 delete(#amqqueue{ pid = QPid }, IfUnused, IfEmpty) ->
-    delegate_call(QPid, {delete, IfUnused, IfEmpty}, infinity).
+    delegate_call(QPid, {delete, IfUnused, IfEmpty}).
 
-purge(#amqqueue{ pid = QPid }) -> delegate_call(QPid, purge, infinity).
+purge(#amqqueue{ pid = QPid }) -> delegate_call(QPid, purge).
 
 deliver(QPid, Delivery = #delivery{immediate = true}) ->
     gen_server2:call(QPid, {deliver_immediately, Delivery}, infinity);
@@ -373,7 +370,7 @@ deliver(QPid, Delivery) ->
     true.
 
 requeue(QPid, MsgIds, ChPid) ->
-    delegate_call(QPid, {requeue, MsgIds, ChPid}, infinity).
+    delegate_call(QPid, {requeue, MsgIds, ChPid}).
 
 ack(QPid, Txn, MsgIds, ChPid) ->
     delegate_cast(QPid, {ack, Txn, MsgIds, ChPid}).
@@ -402,20 +399,18 @@ limit_all(QPids, ChPid, LimiterPid) ->
              end).
 
 basic_get(#amqqueue{pid = QPid}, ChPid, NoAck) ->
-    delegate_call(QPid, {basic_get, ChPid, NoAck}, infinity).
+    delegate_call(QPid, {basic_get, ChPid, NoAck}).
 
 basic_consume(#amqqueue{pid = QPid}, NoAck, ChPid, LimiterPid,
               ConsumerTag, ExclusiveConsume, OkMsg) ->
     delegate_call(QPid, {basic_consume, NoAck, ChPid,
-                         LimiterPid, ConsumerTag, ExclusiveConsume, OkMsg},
-                  infinity).
+                         LimiterPid, ConsumerTag, ExclusiveConsume, OkMsg}).
 
 basic_cancel(#amqqueue{pid = QPid}, ChPid, ConsumerTag, OkMsg) ->
-    ok = delegate_call(QPid, {basic_cancel, ChPid, ConsumerTag, OkMsg},
-                       infinity).
+    ok = delegate_call(QPid, {basic_cancel, ChPid, ConsumerTag, OkMsg}).
 
 notify_sent(QPid, ChPid) ->
-    delegate_cast(QPid, {notify_sent, ChPid}).
+    gen_server2:cast(QPid, {notify_sent, ChPid}).
 
 unblock(QPid, ChPid) ->
     delegate_cast(QPid, {unblock, ChPid}).
@@ -432,17 +427,15 @@ internal_delete1(QueueName) ->
     rabbit_binding:remove_for_destination(QueueName).
 
 internal_delete(QueueName) ->
-    rabbit_misc:execute_mnesia_transaction(
+    rabbit_misc:execute_mnesia_tx_with_tail(
       fun () ->
               case mnesia:wread({rabbit_queue, QueueName}) of
-                  []  -> {error, not_found};
-                  [_] -> internal_delete1(QueueName)
+                  []  -> rabbit_misc:const({error, not_found});
+                  [_] -> Deletions = internal_delete1(QueueName),
+                         fun (Tx) -> ok = rabbit_binding:process_deletions(
+                                            Deletions, Tx)
+                         end
               end
-      end,
-      fun ({error, _} = Err, _Tx) ->
-              Err;
-          (Deletions, Tx) ->
-              ok = rabbit_binding:process_deletions(Deletions, Tx)
       end).
 
 maybe_run_queue_via_backing_queue(QPid, Fun) ->
@@ -505,8 +498,8 @@ safe_delegate_call_ok(F, Pids) ->
         {_, Bad} -> {error, Bad}
     end.
 
-delegate_call(Pid, Msg, Timeout) ->
-    delegate:invoke(Pid, fun (P) -> gen_server2:call(P, Msg, Timeout) end).
+delegate_call(Pid, Msg) ->
+    delegate:invoke(Pid, fun (P) -> gen_server2:call(P, Msg, infinity) end).
 
 delegate_cast(Pid, Msg) ->
     delegate:invoke_no_result(Pid, fun (P) -> gen_server2:cast(P, Msg) end).
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 663977ba87..e794b4aa1e 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -21,7 +21,7 @@
 -behaviour(gen_server2).
 
 -define(UNSENT_MESSAGE_LIMIT,          100).
--define(SYNC_INTERVAL,                 5). %% milliseconds
+-define(SYNC_INTERVAL,                 25). %% milliseconds
 -define(RAM_DURATION_UPDATE_INTERVAL,  5000).
 
 -define(BASE_MESSAGE_PROPERTIES,
@@ -122,6 +122,8 @@ terminate({shutdown, _}, State = #q{backing_queue = BQ}) ->
 terminate(_Reason,       State = #q{backing_queue = BQ}) ->
     %% FIXME: How do we cancel active subscriptions?
     terminate_shutdown(fun (BQS) ->
+                               rabbit_event:notify(
+                                 queue_deleted, [{pid, self()}]),
                                BQS1 = BQ:delete_and_terminate(BQS),
                                %% don't care if the internal delete
                                %% doesn't return 'ok'.
@@ -186,7 +188,6 @@ terminate_shutdown(Fun, State) ->
                               end, BQS, all_ch_record()),
                      [emit_consumer_deleted(Ch, CTag)
                       || {Ch, CTag, _} <- consumers(State1)],
-                     rabbit_event:notify(queue_deleted, [{pid, self()}]),
                      State1#q{backing_queue_state = Fun(BQS1)}
     end.
 
@@ -332,11 +333,6 @@ ch_record_state_transition(OldCR, NewCR) ->
        true                               -> ok
     end.
 
-record_current_channel_tx(ChPid, Txn) ->
-    %% as a side effect this also starts monitoring the channel (if
-    %% that wasn't happening already)
-    store_ch_record((ch_record(ChPid))#cr{txn = Txn}).
-
 deliver_msgs_to_consumers(Funs = {PredFun, DeliverFun}, FunAcc,
                           State = #q{q = #amqqueue{name = QName},
                                      active_consumers = ActiveConsumers,
@@ -495,7 +491,7 @@ attempt_delivery(#delivery{txn = Txn,
                  {NeedsConfirming,
                   State = #q{backing_queue = BQ,
                             backing_queue_state = BQS}}) ->
-    record_current_channel_tx(ChPid, Txn),
+    store_ch_record((ch_record(ChPid))#cr{txn = Txn}),
     {true,
      NeedsConfirming,
      State#q{backing_queue_state =
@@ -591,7 +587,7 @@ handle_ch_down(DownPid, State = #q{exclusive_consumer = Holder}) ->
                 true  -> {stop, State1};
                 false -> State2 = case Txn of
                                       none -> State1;
-                                      _    -> rollback_transaction(Txn, ChPid,
+                                      _    -> rollback_transaction(Txn, C,
                                                                    State1)
                                   end,
                          {ok, requeue_and_run(sets:to_list(ChAckTags),
@@ -622,31 +618,32 @@ maybe_send_reply(ChPid, Msg) -> ok = rabbit_channel:send_command(ChPid, Msg).
 
 qname(#q{q = #amqqueue{name = QName}}) -> QName.
 
+backing_queue_idle_timeout(State = #q{backing_queue = BQ}) ->
+    maybe_run_queue_via_backing_queue(
+      fun (BQS) -> {[], BQ:idle_timeout(BQS)} end, State).
+
 maybe_run_queue_via_backing_queue(Fun, State = #q{backing_queue_state = BQS}) ->
     {Guids, BQS1} = Fun(BQS),
     run_message_queue(
       confirm_messages(Guids, State#q{backing_queue_state = BQS1})).
 
-commit_transaction(Txn, From, ChPid, State = #q{backing_queue       = BQ,
-                                                backing_queue_state = BQS,
-                                                ttl                 = TTL}) ->
-    {AckTags, BQS1} = BQ:tx_commit(Txn,
-                                   fun () -> gen_server2:reply(From, ok) end,
-                                   reset_msg_expiry_fun(TTL),
-                                   BQS),
-    %% ChPid must be known here because of the participant management
-    %% by the channel.
-    C = #cr{acktags = ChAckTags} = lookup_ch(ChPid),
+commit_transaction(Txn, From, C = #cr{acktags = ChAckTags},
+                   State = #q{backing_queue       = BQ,
+                              backing_queue_state = BQS,
+                              ttl                 = TTL}) ->
+    {AckTags, BQS1} = BQ:tx_commit(
+                        Txn, fun () -> gen_server2:reply(From, ok) end,
+                        reset_msg_expiry_fun(TTL), BQS),
     ChAckTags1 = subtract_acks(ChAckTags, AckTags),
     maybe_store_ch_record(C#cr{acktags = ChAckTags1, txn = none}),
     State#q{backing_queue_state = BQS1}.
 
-rollback_transaction(Txn, ChPid, State = #q{backing_queue = BQ,
-                                            backing_queue_state = BQS}) ->
+rollback_transaction(Txn, C, State = #q{backing_queue = BQ,
+                                        backing_queue_state = BQS}) ->
     {_AckTags, BQS1} = BQ:tx_rollback(Txn, BQS),
     %% Iff we removed acktags from the channel record on ack+txn then
-    %% we would add them back in here (would also require ChPid)
-    record_current_channel_tx(ChPid, none),
+    %% we would add them back in here.
+    maybe_store_ch_record(C#cr{txn = none}),
     State#q{backing_queue_state = BQS1}.
 
 subtract_acks(A, B) when is_list(B) ->
@@ -661,13 +658,13 @@ message_properties(#q{ttl=TTL}) ->
     #message_properties{expiry = calculate_msg_expiry(TTL)}.
 
 calculate_msg_expiry(undefined) -> undefined;
-calculate_msg_expiry(TTL)       -> now_millis() + (TTL * 1000).
+calculate_msg_expiry(TTL)       -> now_micros() + (TTL * 1000).
 
 drop_expired_messages(State = #q{ttl = undefined}) ->
     State;
 drop_expired_messages(State = #q{backing_queue_state = BQS,
                                  backing_queue = BQ}) ->
-    Now = now_millis(),
+    Now = now_micros(),
     BQS1 = BQ:dropwhile(
              fun (#message_properties{expiry = Expiry}) ->
                      Now > Expiry
@@ -688,7 +685,7 @@ ensure_ttl_timer(State = #q{backing_queue       = BQ,
 ensure_ttl_timer(State) ->
     State.
 
-now_millis() -> timer:now_diff(now(), {0,0,0}).
+now_micros() -> timer:now_diff(now(), {0,0,0}).
 
 infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
 
@@ -794,20 +791,20 @@ handle_call({init, Recover}, From,
 
 handle_call({init, Recover}, From,
             State = #q{q = #amqqueue{exclusive_owner = Owner}}) ->
-    case rpc:call(node(Owner), erlang, is_process_alive, [Owner]) of
-        true -> erlang:monitor(process, Owner),
-                declare(Recover, From, State);
-        _    -> #q{q = #amqqueue{name = QName, durable = IsDurable},
-                   backing_queue = BQ, backing_queue_state = undefined} = State,
-                gen_server2:reply(From, not_found),
-                case Recover of
-                    true -> ok;
-                    _    -> rabbit_log:warning(
-                              "Queue ~p exclusive owner went away~n", [QName])
-                end,
-                BQS = BQ:init(QName, IsDurable, Recover),
-                %% Rely on terminate to delete the queue.
-                {stop, normal, State#q{backing_queue_state = BQS}}
+    case rabbit_misc:is_process_alive(Owner) of
+        true  -> erlang:monitor(process, Owner),
+                 declare(Recover, From, State);
+        false -> #q{backing_queue = BQ, backing_queue_state = undefined,
+                    q = #amqqueue{name = QName, durable = IsDurable}} = State,
+                 gen_server2:reply(From, not_found),
+                 case Recover of
+                     true -> ok;
+                     _    -> rabbit_log:warning(
+                               "Queue ~p exclusive owner went away~n", [QName])
+                 end,
+                 BQS = BQ:init(QName, IsDurable, Recover),
+                 %% Rely on terminate to delete the queue.
+                 {stop, normal, State#q{backing_queue_state = BQS}}
     end;
 
 handle_call(info, _From, State) ->
@@ -848,8 +845,11 @@ handle_call({deliver, Delivery}, From, State) ->
     noreply(NewState);
 
 handle_call({commit, Txn, ChPid}, From, State) ->
-    NewState = commit_transaction(Txn, From, ChPid, State),
-    noreply(run_message_queue(NewState));
+    case lookup_ch(ChPid) of
+        not_found -> reply(ok, State);
+        C         -> noreply(run_message_queue(
+                               commit_transaction(Txn, From, C, State)))
+    end;
 
 handle_call({notify_down, ChPid}, _From, State) ->
     %% we want to do this synchronously, so that auto_deleted queues
@@ -1001,10 +1001,8 @@ handle_call({maybe_run_queue_via_backing_queue, Fun}, _From, State) ->
 handle_cast({maybe_run_queue_via_backing_queue, Fun}, State) ->
     noreply(maybe_run_queue_via_backing_queue(Fun, State));
 
-handle_cast(sync_timeout, State = #q{backing_queue = BQ,
-                                     backing_queue_state = BQS}) ->
-    noreply(State#q{backing_queue_state = BQ:idle_timeout(BQS),
-                    sync_timer_ref = undefined});
+handle_cast(sync_timeout, State) ->
+    noreply(backing_queue_idle_timeout(State#q{sync_timer_ref = undefined}));
 
 handle_cast({deliver, Delivery}, State) ->
     %% Asynchronous, non-"mandatory", non-"immediate" deliver mode.
@@ -1048,7 +1046,10 @@ handle_cast({reject, AckTags, Requeue, ChPid},
     end;
 
 handle_cast({rollback, Txn, ChPid}, State) ->
-    noreply(rollback_transaction(Txn, ChPid, State));
+    noreply(case lookup_ch(ChPid) of
+                not_found -> State;
+                C         -> rollback_transaction(Txn, C, State)
+            end);
 
 handle_cast(delete_immediately, State) ->
     {stop, normal, State};
@@ -1135,9 +1136,8 @@ handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) ->
         {stop, NewState} -> {stop, normal, NewState}
     end;
 
-handle_info(timeout, State = #q{backing_queue = BQ}) ->
-    noreply(maybe_run_queue_via_backing_queue(
-              fun (BQS) -> {[], BQ:idle_timeout(BQS)} end, State));
+handle_info(timeout, State) ->
+    noreply(backing_queue_idle_timeout(State));
 
 handle_info({'EXIT', _Pid, Reason}, State) ->
     {stop, Reason, State};
@@ -1151,15 +1151,15 @@ handle_pre_hibernate(State = #q{backing_queue_state = undefined}) ->
 handle_pre_hibernate(State = #q{backing_queue = BQ,
                                 backing_queue_state = BQS,
                                 stats_timer = StatsTimer}) ->
-    BQS1 = BQ:handle_pre_hibernate(BQS),
-    %% no activity for a while == 0 egress and ingress rates
+    {RamDuration, BQS1} = BQ:ram_duration(BQS),
     DesiredDuration =
-        rabbit_memory_monitor:report_ram_duration(self(), infinity),
+        rabbit_memory_monitor:report_ram_duration(self(), RamDuration),
     BQS2 = BQ:set_ram_duration_target(DesiredDuration, BQS1),
+    BQS3 = BQ:handle_pre_hibernate(BQS2),
     rabbit_event:if_enabled(StatsTimer,
                             fun () ->
                                     emit_stats(State, [{idle_since, now()}])
                             end),
     State1 = State#q{stats_timer = rabbit_event:stop_stats_timer(StatsTimer),
-                     backing_queue_state = BQS2},
+                     backing_queue_state = BQS3},
     {hibernate, stop_rate_timer(State1)}.
diff --git a/src/rabbit_auth_mechanism_plain.erl b/src/rabbit_auth_mechanism_plain.erl
index 7d9dcd2007..1ca07018e4 100644
--- a/src/rabbit_auth_mechanism_plain.erl
+++ b/src/rabbit_auth_mechanism_plain.erl
@@ -33,6 +33,10 @@
 %% SASL PLAIN, as used by the Qpid Java client and our clients. Also,
 %% apparently, by OpenAMQ.
 
+%% TODO: once the minimum erlang becomes R13B03, reimplement this
+%% using the binary module - that makes use of BIFs to do binary
+%% matching and will thus be much faster.
+
 description() ->
     [{name, <<"PLAIN">>},
      {description, <<"SASL PLAIN authentication mechanism">>}].
@@ -41,11 +45,32 @@ init(_Sock) ->
     [].
 
 handle_response(Response, _State) ->
-    %% The '%%"' at the end of the next line is for Emacs
-    case re:run(Response, "^\\0([^\\0]*)\\0([^\\0]*)$",%%"
-                [{capture, all_but_first, binary}]) of
-        {match, [User, Pass]} ->
+    case extract_user_pass(Response) of
+        {ok, User, Pass} ->
             rabbit_access_control:check_user_pass_login(User, Pass);
-        _ ->
+        error ->
             {protocol_error, "response ~p invalid", [Response]}
     end.
+
+extract_user_pass(Response) ->
+    case extract_elem(Response) of
+        {ok, User, Response1} -> case extract_elem(Response1) of
+                                     {ok, Pass, <<>>} -> {ok, User, Pass};
+                                     _                -> error
+                                 end;
+        error                 -> error
+    end.
+
+extract_elem(<<0:8, Rest/binary>>) ->
+    Count = next_null_pos(Rest),
+    <<Elem:Count/binary, Rest1/binary>> = Rest,
+    {ok, Elem, Rest1};
+extract_elem(_) ->
+    error.
+
+next_null_pos(Bin) ->
+    next_null_pos(Bin, 0).
+
+next_null_pos(<<>>, Count)                  -> Count;
+next_null_pos(<<0:8, _Rest/binary>>, Count) -> Count;
+next_null_pos(<<_:8, Rest/binary>>,  Count) -> next_null_pos(Rest, Count + 1).
diff --git a/src/rabbit_binary_generator.erl b/src/rabbit_binary_generator.erl
index d67c7f58e1..dc81ace6bf 100644
--- a/src/rabbit_binary_generator.erl
+++ b/src/rabbit_binary_generator.erl
@@ -61,8 +61,7 @@
 -spec(map_exception/3 :: (rabbit_channel:channel_number(),
                           rabbit_types:amqp_error() | any(),
                           rabbit_types:protocol()) ->
-                              {boolean(),
-                               rabbit_channel:channel_number(),
+                              {rabbit_channel:channel_number(),
                                rabbit_framing:amqp_method_record()}).
 
 -endif.
@@ -301,24 +300,21 @@ clear_encoded_content(Content = #content{}) ->
 map_exception(Channel, Reason, Protocol) ->
     {SuggestedClose, ReplyCode, ReplyText, FailedMethod} =
         lookup_amqp_exception(Reason, Protocol),
-    ShouldClose = SuggestedClose orelse (Channel == 0),
     {ClassId, MethodId} = case FailedMethod of
                               {_, _} -> FailedMethod;
                               none   -> {0, 0};
                               _      -> Protocol:method_id(FailedMethod)
                           end,
-    {CloseChannel, CloseMethod} =
-        case ShouldClose of
-            true  -> {0, #'connection.close'{reply_code = ReplyCode,
-                                             reply_text = ReplyText,
-                                             class_id = ClassId,
-                                             method_id = MethodId}};
-            false -> {Channel, #'channel.close'{reply_code = ReplyCode,
-                                                reply_text = ReplyText,
-                                                class_id = ClassId,
-                                                method_id = MethodId}}
-        end,
-    {ShouldClose, CloseChannel, CloseMethod}.
+    case SuggestedClose orelse (Channel == 0) of
+        true  -> {0, #'connection.close'{reply_code = ReplyCode,
+                                         reply_text = ReplyText,
+                                         class_id   = ClassId,
+                                         method_id  = MethodId}};
+        false -> {Channel, #'channel.close'{reply_code = ReplyCode,
+                                            reply_text = ReplyText,
+                                            class_id   = ClassId,
+                                            method_id  = MethodId}}
+    end.
 
 lookup_amqp_exception(#amqp_error{name        = Name,
                                   explanation = Expl,
diff --git a/src/rabbit_binding.erl b/src/rabbit_binding.erl
index b270927be3..96a22dcaf1 100644
--- a/src/rabbit_binding.erl
+++ b/src/rabbit_binding.erl
@@ -331,7 +331,7 @@ group_bindings_fold(Fun, SrcName, Acc, Removed, Bindings) ->
     group_bindings_fold(Fun, Fun(SrcName, Bindings, Acc), Removed).
 
 maybe_auto_delete(XName, Bindings, Deletions) ->
-    case mnesia:read(rabbit_exchange, XName) of
+    case mnesia:read({rabbit_exchange, XName}) of
         [] ->
             add_deletion(XName, {undefined, not_deleted, Bindings}, Deletions);
         [X] ->
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index b92206ad4e..34a5e5a440 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -20,29 +20,32 @@
 
 -behaviour(gen_server2).
 
--export([start_link/7, do/2, do/3, flush/1, shutdown/1]).
+-export([start_link/9, do/2, do/3, flush/1, shutdown/1]).
 -export([send_command/2, deliver/4, flushed/2, confirm/2]).
 -export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1]).
--export([emit_stats/1]).
+-export([emit_stats/1, ready_for_close/1]).
 
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
          handle_info/2, handle_pre_hibernate/1, prioritise_call/3,
          prioritise_cast/2]).
 
--record(ch, {state, channel, reader_pid, writer_pid, limiter_pid,
+-record(ch, {state, protocol, channel, reader_pid, writer_pid, limiter_pid,
              start_limiter_fun, transaction_id, tx_participants, next_tag,
              uncommitted_ack_q, unacked_message_q,
              user, virtual_host, most_recently_declared_queue,
              consumer_mapping, blocking, queue_collector_pid, stats_timer,
-             confirm_enabled, publish_seqno, unconfirmed, confirmed}).
+             confirm_enabled, publish_seqno, unconfirmed_mq, unconfirmed_qm,
+             confirmed, capabilities}).
 
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 
 -define(STATISTICS_KEYS,
         [pid,
          transactional,
+         confirm,
          consumer_count,
          messages_unacknowledged,
+         messages_unconfirmed,
          acks_uncommitted,
          prefetch_count,
          client_flow_blocked]).
@@ -64,10 +67,10 @@
 
 -type(channel_number() :: non_neg_integer()).
 
--spec(start_link/7 ::
-      (channel_number(), pid(), pid(), rabbit_types:user(),
-       rabbit_types:vhost(), pid(),
-       fun ((non_neg_integer()) -> rabbit_types:ok(pid()))) ->
+-spec(start_link/9 ::
+      (channel_number(), pid(), pid(), rabbit_types:protocol(),
+       rabbit_types:user(), rabbit_types:vhost(), rabbit_framing:amqp_table(),
+       pid(), fun ((non_neg_integer()) -> rabbit_types:ok(pid()))) ->
                            rabbit_types:ok_pid_or_error()).
 -spec(do/2 :: (pid(), rabbit_framing:amqp_method_record()) -> 'ok').
 -spec(do/3 :: (pid(), rabbit_framing:amqp_method_record(),
@@ -87,15 +90,17 @@
 -spec(info_all/0 :: () -> [rabbit_types:infos()]).
 -spec(info_all/1 :: (rabbit_types:info_keys()) -> [rabbit_types:infos()]).
 -spec(emit_stats/1 :: (pid()) -> 'ok').
+-spec(ready_for_close/1 :: (pid()) -> 'ok').
 
 -endif.
 
 %%----------------------------------------------------------------------------
 
-start_link(Channel, ReaderPid, WriterPid, User, VHost, CollectorPid,
-           StartLimiterFun) ->
-    gen_server2:start_link(?MODULE, [Channel, ReaderPid, WriterPid, User,
-                                     VHost, CollectorPid, StartLimiterFun], []).
+start_link(Channel, ReaderPid, WriterPid, Protocol, User, VHost, Capabilities,
+           CollectorPid, StartLimiterFun) ->
+    gen_server2:start_link(
+      ?MODULE, [Channel, ReaderPid, WriterPid, Protocol, User, VHost,
+                Capabilities, CollectorPid, StartLimiterFun], []).
 
 do(Pid, Method) ->
     do(Pid, Method, none).
@@ -104,7 +109,7 @@ do(Pid, Method, Content) ->
     gen_server2:cast(Pid, {method, Method, Content}).
 
 flush(Pid) ->
-    gen_server2:call(Pid, flush).
+    gen_server2:call(Pid, flush, infinity).
 
 shutdown(Pid) ->
     gen_server2:cast(Pid, terminate).
@@ -144,14 +149,18 @@ info_all(Items) ->
 emit_stats(Pid) ->
     gen_server2:cast(Pid, emit_stats).
 
+ready_for_close(Pid) ->
+    gen_server2:cast(Pid, ready_for_close).
+
 %%---------------------------------------------------------------------------
 
-init([Channel, ReaderPid, WriterPid, User, VHost, CollectorPid,
-      StartLimiterFun]) ->
+init([Channel, ReaderPid, WriterPid, Protocol, User, VHost, Capabilities,
+      CollectorPid, StartLimiterFun]) ->
     process_flag(trap_exit, true),
     ok = pg_local:join(rabbit_channels, self()),
     StatsTimer = rabbit_event:init_stats_timer(),
     State = #ch{state                   = starting,
+                protocol                = Protocol,
                 channel                 = Channel,
                 reader_pid              = ReaderPid,
                 writer_pid              = WriterPid,
@@ -171,8 +180,10 @@ init([Channel, ReaderPid, WriterPid, User, VHost, CollectorPid,
                 stats_timer             = StatsTimer,
                 confirm_enabled         = false,
                 publish_seqno           = 1,
-                unconfirmed             = gb_trees:empty(),
-                confirmed               = []},
+                unconfirmed_mq          = gb_trees:empty(),
+                unconfirmed_qm          = gb_trees:empty(),
+                confirmed               = [],
+                capabilities            = Capabilities},
     rabbit_event:notify(channel_created, infos(?CREATION_EVENT_KEYS, State)),
     rabbit_event:if_enabled(StatsTimer,
                             fun() -> internal_emit_stats(State) end),
@@ -216,14 +227,11 @@ handle_cast({method, Method, Content}, State) ->
         {noreply, NewState} ->
             noreply(NewState);
         stop ->
-            {stop, normal, State#ch{state = terminating}}
+            {stop, normal, State}
     catch
         exit:Reason = #amqp_error{} ->
             MethodName = rabbit_misc:method_record_type(Method),
-            {stop, normal, terminating(Reason#amqp_error{method = MethodName},
-                                       State)};
-        exit:normal ->
-            {stop, normal, State};
+            send_exception(Reason#amqp_error{method = MethodName}, State);
         _:Reason ->
             {stop, {Reason, erlang:get_stacktrace()}, State}
     end;
@@ -231,6 +239,11 @@ handle_cast({method, Method, Content}, State) ->
 handle_cast({flushed, QPid}, State) ->
     {noreply, queue_blocked(QPid, State), hibernate};
 
+handle_cast(ready_for_close, State = #ch{state      = closing,
+                                         writer_pid = WriterPid}) ->
+    ok = rabbit_writer:send_command_sync(WriterPid, #'channel.close_ok'{}),
+    {stop, normal, State};
+
 handle_cast(terminate, State) ->
     {stop, normal, State};
 
@@ -275,17 +288,23 @@ handle_cast({confirm, MsgSeqNos, From}, State) ->
 handle_info(timeout, State) ->
     noreply(State);
 
-handle_info({'DOWN', _MRef, process, QPid, _Reason},
-            State = #ch{unconfirmed = UC}) ->
-    %% TODO: this does a complete scan and partial rebuild of the
-    %% tree, which is quite efficient. To do better we'd need to
-    %% maintain a secondary mapping, from QPids to MsgSeqNos.
-    {MsgSeqNos, UC1} = remove_queue_unconfirmed(
-                         gb_trees:next(gb_trees:iterator(UC)), QPid,
-                         {[], UC}),
+handle_info({'DOWN', _MRef, process, QPid, Reason},
+            State = #ch{unconfirmed_qm = UQM}) ->
+    MsgSeqNos = case gb_trees:lookup(QPid, UQM) of
+                    {value, MsgSet} -> gb_sets:to_list(MsgSet);
+                    none            -> []
+                end,
+    %% We remove the MsgSeqNos from UQM before calling
+    %% process_confirms to prevent each MsgSeqNo being removed from
+    %% the set one by one which which would be inefficient
+    State1 = State#ch{unconfirmed_qm = gb_trees:delete_any(QPid, UQM)},
+    {MXs, State2} = process_confirms(MsgSeqNos, QPid, State1),
     erase_queue_stats(QPid),
-    noreply(queue_blocked(QPid, record_confirms(MsgSeqNos,
-                                                State#ch{unconfirmed = UC1}))).
+    State3 = (case Reason of
+                 normal -> fun record_confirms/2;
+                 _      -> fun send_nacks/2
+              end)(MXs, State2),
+    noreply(queue_blocked(QPid, State3)).
 
 handle_pre_hibernate(State = #ch{stats_timer = StatsTimer}) ->
     ok = clear_permission_cache(),
@@ -297,18 +316,16 @@ handle_pre_hibernate(State = #ch{stats_timer = StatsTimer}) ->
     StatsTimer1 = rabbit_event:stop_stats_timer(StatsTimer),
     {hibernate, State#ch{stats_timer = StatsTimer1}}.
 
-terminate(_Reason, State = #ch{state = terminating}) ->
-    terminate(State);
-
 terminate(Reason, State) ->
-    Res = rollback_and_notify(State),
+    {Res, _State1} = rollback_and_notify(State),
     case Reason of
         normal            -> ok = Res;
         shutdown          -> ok = Res;
         {shutdown, _Term} -> ok = Res;
         _                 -> ok
     end,
-    terminate(State).
+    pg_local:leave(rabbit_channels, self()),
+    rabbit_event:notify(channel_closed, [{pid, self()}]).
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
@@ -346,10 +363,22 @@ return_ok(State, false, Msg)  -> {reply, Msg, State}.
 ok_msg(true, _Msg) -> undefined;
 ok_msg(false, Msg) -> Msg.
 
-terminating(Reason, State = #ch{channel = Channel, reader_pid = Reader}) ->
-    ok = rollback_and_notify(State),
-    Reader ! {channel_exit, Channel, Reason},
-    State#ch{state = terminating}.
+send_exception(Reason, State = #ch{protocol   = Protocol,
+                                   channel    = Channel,
+                                   writer_pid = WriterPid,
+                                   reader_pid = ReaderPid}) ->
+    {CloseChannel, CloseMethod} =
+        rabbit_binary_generator:map_exception(Channel, Reason, Protocol),
+    rabbit_log:error("connection ~p, channel ~p - error:~n~p~n",
+                     [ReaderPid, Channel, Reason]),
+    %% something bad's happened: rollback_and_notify may not be 'ok'
+    {_Result, State1} = rollback_and_notify(State),
+    case CloseChannel of
+        Channel -> ok = rabbit_writer:send_command(WriterPid, CloseMethod),
+                   {noreply, State1};
+        _       -> ReaderPid ! {channel_exit, Channel, Reason},
+                   {stop, normal, State1}
+    end.
 
 return_queue_declare_ok(#resource{name = ActualName},
                         NoWait, MessageCount, ConsumerCount, State) ->
@@ -471,38 +500,55 @@ queue_blocked(QPid, State = #ch{blocking = Blocking}) ->
                       State#ch{blocking = Blocking1}
     end.
 
-remove_queue_unconfirmed(none, _QPid, Acc) ->
-    Acc;
-remove_queue_unconfirmed({MsgSeqNo, Qs, Next}, QPid, Acc) ->
-    remove_queue_unconfirmed(gb_trees:next(Next), QPid,
-                             remove_qmsg(MsgSeqNo, QPid, Qs, Acc)).
-
-record_confirm(undefined, State) -> State;
-record_confirm(MsgSeqNo,  State) -> record_confirms([MsgSeqNo], State).
+record_confirm(undefined, _, State) ->
+    State;
+record_confirm(MsgSeqNo, XName, State) ->
+    record_confirms([{MsgSeqNo, XName}], State).
 
 record_confirms([], State) ->
     State;
-record_confirms(MsgSeqNos, State = #ch{confirmed = C}) ->
-    State#ch{confirmed = [MsgSeqNos | C]}.
+record_confirms(MXs, State = #ch{confirmed = C}) ->
+    State#ch{confirmed = [MXs | C]}.
 
 confirm([], _QPid, State) ->
     State;
-confirm(MsgSeqNos, QPid, State = #ch{unconfirmed = UC}) ->
-    {DoneMessages, UC2} =
+confirm(MsgSeqNos, QPid, State) ->
+    {MXs, State1} = process_confirms(MsgSeqNos, QPid, State),
+    record_confirms(MXs, State1).
+
+process_confirms(MsgSeqNos, QPid, State = #ch{unconfirmed_mq = UMQ,
+                                              unconfirmed_qm = UQM}) ->
+    {MXs, UMQ1, UQM1} =
         lists:foldl(
-          fun(MsgSeqNo, {_DMs, UC0} = Acc) ->
-                  case gb_trees:lookup(MsgSeqNo, UC0) of
-                      none        -> Acc;
-                      {value, Qs} -> remove_qmsg(MsgSeqNo, QPid, Qs, Acc)
+          fun(MsgSeqNo, {_DMs, UMQ0, _UQM} = Acc) ->
+                  case gb_trees:lookup(MsgSeqNo, UMQ0) of
+                      {value, XQ} -> remove_unconfirmed(MsgSeqNo, QPid, XQ, Acc,
+                                                        State);
+                      none        -> Acc
                   end
-          end, {[], UC}, MsgSeqNos),
-    record_confirms(DoneMessages, State#ch{unconfirmed = UC2}).
-
-remove_qmsg(MsgSeqNo, QPid, Qs, {MsgSeqNos, UC}) ->
-    Qs1 = sets:del_element(QPid, Qs),
-    case sets:size(Qs1) of
-        0 -> {[MsgSeqNo | MsgSeqNos], gb_trees:delete(MsgSeqNo, UC)};
-        _ -> {MsgSeqNos, gb_trees:update(MsgSeqNo, Qs1, UC)}
+          end, {[], UMQ, UQM}, MsgSeqNos),
+    {MXs, State#ch{unconfirmed_mq = UMQ1, unconfirmed_qm = UQM1}}.
+
+remove_unconfirmed(MsgSeqNo, QPid, {XName, Qs}, {MXs, UMQ, UQM}, State) ->
+    %% these confirms will be emitted even when a queue dies, but that
+    %% should be fine, since the queue stats get erased immediately
+    maybe_incr_stats([{{QPid, XName}, 1}], confirm, State),
+    UQM1 = case gb_trees:lookup(QPid, UQM) of
+               {value, MsgSeqNos} ->
+                   MsgSeqNos1 = gb_sets:delete(MsgSeqNo, MsgSeqNos),
+                   case gb_sets:is_empty(MsgSeqNos1) of
+                       true  -> gb_trees:delete(QPid, UQM);
+                       false -> gb_trees:update(QPid, MsgSeqNos1, UQM)
+                   end;
+               none ->
+                   UQM
+           end,
+    Qs1 = gb_sets:del_element(QPid, Qs),
+    case gb_sets:is_empty(Qs1) of
+        true ->
+            {[{MsgSeqNo, XName} | MXs], gb_trees:delete(MsgSeqNo, UMQ), UQM1};
+        false ->
+            {MXs, gb_trees:update(MsgSeqNo, {XName, Qs1}, UMQ), UQM1}
     end.
 
 handle_method(#'channel.open'{}, _, State = #ch{state = starting}) ->
@@ -515,11 +561,20 @@ handle_method(#'channel.open'{}, _, _State) ->
 handle_method(_Method, _, #ch{state = starting}) ->
     rabbit_misc:protocol_error(channel_error, "expected 'channel.open'", []);
 
-handle_method(#'channel.close'{}, _, State = #ch{writer_pid = WriterPid}) ->
-    ok = rollback_and_notify(State),
-    ok = rabbit_writer:send_command_sync(WriterPid, #'channel.close_ok'{}),
+handle_method(#'channel.close_ok'{}, _, #ch{state = closing}) ->
     stop;
 
+handle_method(#'channel.close'{}, _, State = #ch{state = closing}) ->
+    {reply, #'channel.close_ok'{}, State};
+
+handle_method(_Method, _, State = #ch{state = closing}) ->
+    {noreply, State};
+
+handle_method(#'channel.close'{}, _, State = #ch{reader_pid = ReaderPid}) ->
+    {ok, State1} = rollback_and_notify(State),
+    ReaderPid ! {channel_closing, self()},
+    {noreply, State1};
+
 handle_method(#'access.request'{},_, State) ->
     {reply, #'access.request_ok'{ticket = 1}, State};
 
@@ -555,7 +610,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
           Exchange,
           rabbit_basic:delivery(Mandatory, Immediate, TxnKey, Message,
                                 MsgSeqNo)),
-    State2 = process_routing_result(RoutingRes, DeliveredQPids,
+    State2 = process_routing_result(RoutingRes, DeliveredQPids, ExchangeName,
                                     MsgSeqNo, Message, State1),
     maybe_incr_stats([{ExchangeName, 1} |
                       [{{QPid, ExchangeName}, 1} ||
@@ -729,16 +784,22 @@ handle_method(#'basic.qos'{prefetch_count = PrefetchCount},
     {reply, #'basic.qos_ok'{}, State#ch{limiter_pid = LimiterPid2}};
 
 handle_method(#'basic.recover_async'{requeue = true},
-              _, State = #ch{unacked_message_q = UAMQ}) ->
+              _, State = #ch{unacked_message_q = UAMQ,
+                             limiter_pid = LimiterPid}) ->
+    OkFun = fun () -> ok end,
     ok = fold_per_queue(
            fun (QPid, MsgIds, ok) ->
                    %% The Qpid python test suite incorrectly assumes
                    %% that messages will be requeued in their original
                    %% order. To keep it happy we reverse the id list
                    %% since we are given them in reverse order.
-                   rabbit_amqqueue:requeue(
-                     QPid, lists:reverse(MsgIds), self())
+                   rabbit_misc:with_exit_handler(
+                     OkFun, fun () ->
+                                    rabbit_amqqueue:requeue(
+                                      QPid, lists:reverse(MsgIds), self())
+                            end)
            end, ok, UAMQ),
+    ok = notify_limiter(LimiterPid, UAMQ),
     %% No answer required - basic.recover is the newer, synchronous
     %% variant of this method
     {noreply, State#ch{unacked_message_q = queue:new()}};
@@ -1064,9 +1125,8 @@ binding_action(Fun, ExchangeNameBin, DestinationType, DestinationNameBin,
 basic_return(#basic_message{exchange_name = ExchangeName,
                             routing_key   = RoutingKey,
                             content       = Content},
-             WriterPid, Reason) ->
-    {_Close, ReplyCode, ReplyText} =
-        rabbit_framing_amqp_0_9_1:lookup_amqp_exception(Reason),
+             #ch{protocol = Protocol, writer_pid = WriterPid}, Reason) ->
+    {_Close, ReplyCode, ReplyText} = Protocol:lookup_amqp_exception(Reason),
     ok = rabbit_writer:send_command(
            WriterPid,
            #'basic.return'{reply_code  = ReplyCode,
@@ -1153,10 +1213,13 @@ internal_rollback(State = #ch{transaction_id = TxnKey,
     NewUAMQ = queue:join(UAQ, UAMQ),
     new_tx(State#ch{unacked_message_q = NewUAMQ}).
 
+rollback_and_notify(State = #ch{state = closing}) ->
+    {ok, State};
 rollback_and_notify(State = #ch{transaction_id = none}) ->
-    notify_queues(State);
+    {notify_queues(State), State#ch{state = closing}};
 rollback_and_notify(State) ->
-    notify_queues(internal_rollback(State)).
+    State1 = internal_rollback(State),
+    {notify_queues(State1), State1#ch{state = closing}}.
 
 fold_per_queue(F, Acc0, UAQ) ->
     D = rabbit_misc:queue_fold(
@@ -1222,59 +1285,83 @@ is_message_persistent(Content) ->
             IsPersistent
     end.
 
-process_routing_result(unroutable,    _, MsgSeqNo, Message, State) ->
-    ok = basic_return(Message, State#ch.writer_pid, no_route),
-    record_confirm(MsgSeqNo, State);
-process_routing_result(not_delivered, _, MsgSeqNo, Message, State) ->
-    ok = basic_return(Message, State#ch.writer_pid, no_consumers),
-    record_confirm(MsgSeqNo, State);
-process_routing_result(routed,       [], MsgSeqNo,       _, State) ->
-    record_confirm(MsgSeqNo, State);
-process_routing_result(routed,        _, undefined,      _, State) ->
+process_routing_result(unroutable,    _, XName,  MsgSeqNo, Msg, State) ->
+    ok = basic_return(Msg, State, no_route),
+    record_confirm(MsgSeqNo, XName, State);
+process_routing_result(not_delivered, _, XName,  MsgSeqNo, Msg, State) ->
+    ok = basic_return(Msg, State, no_consumers),
+    record_confirm(MsgSeqNo, XName, State);
+process_routing_result(routed,       [], XName,  MsgSeqNo,   _, State) ->
+    record_confirm(MsgSeqNo, XName, State);
+process_routing_result(routed,        _,     _, undefined,   _, State) ->
     State;
-process_routing_result(routed,    QPids, MsgSeqNo,       _, State) ->
-    #ch{unconfirmed = UC} = State,
-    [maybe_monitor(QPid) || QPid <- QPids],
-    UC1 = gb_trees:insert(MsgSeqNo, sets:from_list(QPids), UC),
-    State#ch{unconfirmed = UC1}.
+process_routing_result(routed,    QPids, XName,  MsgSeqNo,   _, State) ->
+    #ch{unconfirmed_mq = UMQ, unconfirmed_qm = UQM} = State,
+    UMQ1 = gb_trees:insert(MsgSeqNo, {XName, gb_sets:from_list(QPids)}, UMQ),
+    SingletonSet = gb_sets:singleton(MsgSeqNo),
+    UQM1 = lists:foldl(
+             fun (QPid, UQM2) ->
+                     maybe_monitor(QPid),
+                     case gb_trees:lookup(QPid, UQM2) of
+                         {value, MsgSeqNos} ->
+                             MsgSeqNos1 = gb_sets:insert(MsgSeqNo, MsgSeqNos),
+                             gb_trees:update(QPid, MsgSeqNos1, UQM2);
+                         none ->
+                             gb_trees:insert(QPid, SingletonSet, UQM2)
+                     end
+             end, UQM, QPids),
+    State#ch{unconfirmed_mq = UMQ1, unconfirmed_qm = UQM1}.
 
 lock_message(true, MsgStruct, State = #ch{unacked_message_q = UAMQ}) ->
     State#ch{unacked_message_q = queue:in(MsgStruct, UAMQ)};
 lock_message(false, _MsgStruct, State) ->
     State.
 
-send_confirms(State = #ch{confirmed = C}) ->
-    send_confirms(lists:append(C), State #ch{confirmed = []}).
+send_nacks([], State) ->
+    State;
+send_nacks(MXs, State) ->
+    MsgSeqNos = [ MsgSeqNo || {MsgSeqNo, _} <- MXs ],
+    coalesce_and_send(MsgSeqNos,
+                      fun(MsgSeqNo, Multiple) ->
+                              #'basic.nack'{delivery_tag = MsgSeqNo,
+                                            multiple = Multiple}
+                      end, State).
 
+send_confirms(State = #ch{confirmed = C}) ->
+    C1 = lists:append(C),
+    MsgSeqNos = [ begin maybe_incr_stats([{ExchangeName, 1}], confirm, State),
+                        MsgSeqNo
+                  end || {MsgSeqNo, ExchangeName} <- C1 ],
+    send_confirms(MsgSeqNos, State #ch{confirmed = []}).
 send_confirms([], State) ->
     State;
 send_confirms([MsgSeqNo], State = #ch{writer_pid = WriterPid}) ->
-    send_confirm(MsgSeqNo, WriterPid),
+    ok = rabbit_writer:send_command(WriterPid,
+                                    #'basic.ack'{delivery_tag = MsgSeqNo}),
     State;
-send_confirms(Cs, State = #ch{writer_pid  = WriterPid, unconfirmed = UC}) ->
-    SCs = lists:usort(Cs),
-    CutOff = case gb_trees:is_empty(UC) of
-                 true  -> lists:last(SCs) + 1;
-                 false -> {SeqNo, _Qs} = gb_trees:smallest(UC), SeqNo
+send_confirms(Cs, State) ->
+    coalesce_and_send(Cs, fun(MsgSeqNo, Multiple) ->
+                                  #'basic.ack'{delivery_tag = MsgSeqNo,
+                                               multiple     = Multiple}
+                          end, State).
+
+coalesce_and_send(MsgSeqNos, MkMsgFun,
+                  State = #ch{writer_pid = WriterPid, unconfirmed_mq = UMQ}) ->
+    SMsgSeqNos = lists:usort(MsgSeqNos),
+    CutOff = case gb_trees:is_empty(UMQ) of
+                 true  -> lists:last(SMsgSeqNos) + 1;
+                 false -> {SeqNo, _XQ} = gb_trees:smallest(UMQ), SeqNo
              end,
-    {Ms, Ss} = lists:splitwith(fun(X) -> X < CutOff end, SCs),
+    {Ms, Ss} = lists:splitwith(fun(X) -> X < CutOff end, SMsgSeqNos),
     case Ms of
         [] -> ok;
         _  -> ok = rabbit_writer:send_command(
-                     WriterPid, #'basic.ack'{delivery_tag = lists:last(Ms),
-                                             multiple = true})
+                     WriterPid, MkMsgFun(lists:last(Ms), true))
     end,
-    [ok = send_confirm(SeqNo, WriterPid) || SeqNo <- Ss],
+    [ok = rabbit_writer:send_command(
+            WriterPid, MkMsgFun(SeqNo, false)) || SeqNo <- Ss],
     State.
 
-send_confirm(SeqNo, WriterPid) ->
-    ok = rabbit_writer:send_command(WriterPid,
-                                    #'basic.ack'{delivery_tag = SeqNo}).
-
-terminate(_State) ->
-    pg_local:leave(rabbit_channels, self()),
-    rabbit_event:notify(channel_closed, [{pid, self()}]).
-
 infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
 
 i(pid,            _)                                 -> self();
@@ -1283,8 +1370,11 @@ i(number,         #ch{channel          = Channel})   -> Channel;
 i(user,           #ch{user             = User})      -> User#user.username;
 i(vhost,          #ch{virtual_host     = VHost})     -> VHost;
 i(transactional,  #ch{transaction_id   = TxnKey})    -> TxnKey =/= none;
+i(confirm,        #ch{confirm_enabled  = CE})        -> CE;
 i(consumer_count, #ch{consumer_mapping = ConsumerMapping}) ->
     dict:size(ConsumerMapping);
+i(messages_unconfirmed, #ch{unconfirmed_mq = UMQ}) ->
+    gb_trees:size(UMQ);
 i(messages_unacknowledged, #ch{unacked_message_q = UAMQ,
                                uncommitted_ack_q = UAQ}) ->
     queue:len(UAMQ) + queue:len(UAQ);
diff --git a/src/rabbit_channel_sup.erl b/src/rabbit_channel_sup.erl
index d426d55df5..9cc407bc34 100644
--- a/src/rabbit_channel_sup.erl
+++ b/src/rabbit_channel_sup.erl
@@ -31,9 +31,13 @@
 -export_type([start_link_args/0]).
 
 -type(start_link_args() ::
-        {rabbit_types:protocol(), rabbit_net:socket(),
-         rabbit_channel:channel_number(), non_neg_integer(), pid(),
-         rabbit_types:user(), rabbit_types:vhost(), pid()}).
+        {'tcp', rabbit_net:socket(), rabbit_channel:channel_number(),
+         non_neg_integer(), pid(), rabbit_types:protocol(), rabbit_types:user(),
+         rabbit_types:vhost(), rabbit_framing:amqp_table(),
+         pid()} |
+        {'direct', rabbit_channel:channel_number(), pid(),
+         rabbit_types:protocol(), rabbit_types:user(), rabbit_types:vhost(),
+         rabbit_framing:amqp_table(), pid()}).
 
 -spec(start_link/1 :: (start_link_args()) -> {'ok', pid(), {pid(), any()}}).
 
@@ -41,8 +45,8 @@
 
 %%----------------------------------------------------------------------------
 
-start_link({Protocol, Sock, Channel, FrameMax, ReaderPid, User, VHost,
-            Collector}) ->
+start_link({tcp, Sock, Channel, FrameMax, ReaderPid, Protocol, User, VHost,
+            Capabilities, Collector}) ->
     {ok, SupPid} = supervisor2:start_link(?MODULE, []),
     {ok, WriterPid} =
         supervisor2:start_child(
@@ -54,11 +58,23 @@ start_link({Protocol, Sock, Channel, FrameMax, ReaderPid, User, VHost,
         supervisor2:start_child(
           SupPid,
           {channel, {rabbit_channel, start_link,
-                     [Channel, ReaderPid, WriterPid, User, VHost,
-                      Collector, start_limiter_fun(SupPid)]},
+                     [Channel, ReaderPid, WriterPid, Protocol, User, VHost,
+                      Capabilities, Collector, start_limiter_fun(SupPid)]},
            intrinsic, ?MAX_WAIT, worker, [rabbit_channel]}),
     {ok, AState} = rabbit_command_assembler:init(Protocol),
-    {ok, SupPid, {ChannelPid, AState}}.
+    {ok, SupPid, {ChannelPid, AState}};
+start_link({direct, Channel, ClientChannelPid, Protocol, User, VHost,
+            Capabilities, Collector}) ->
+    {ok, SupPid} = supervisor2:start_link(?MODULE, []),
+    {ok, ChannelPid} =
+        supervisor2:start_child(
+            SupPid,
+            {channel, {rabbit_channel, start_link,
+                       [Channel, ClientChannelPid, ClientChannelPid, Protocol,
+                        User, VHost, Capabilities, Collector,
+                        start_limiter_fun(SupPid)]},
+             intrinsic, ?MAX_WAIT, worker, [rabbit_channel]}),
+    {ok, SupPid, {ChannelPid, none}}.
 
 %%----------------------------------------------------------------------------
 
diff --git a/src/tcp_client_sup.erl b/src/rabbit_client_sup.erl
index 1c2bbb6548..dbdc6cd429 100644
--- a/src/tcp_client_sup.erl
+++ b/src/rabbit_client_sup.erl
@@ -14,7 +14,7 @@
 %% Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
 %%
 
--module(tcp_client_sup).
+-module(rabbit_client_sup).
 
 -behaviour(supervisor2).
 
@@ -22,6 +22,21 @@
 
 -export([init/1]).
 
+-include("rabbit.hrl").
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(start_link/1 :: (mfa()) ->
+                          rabbit_types:ok_pid_or_error()).
+-spec(start_link/2 :: ({'local', atom()}, mfa()) ->
+                          rabbit_types:ok_pid_or_error()).
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
 start_link(Callback) ->
     supervisor2:start_link(?MODULE, Callback).
 
@@ -29,6 +44,5 @@ start_link(SupName, Callback) ->
     supervisor2:start_link(SupName, ?MODULE, Callback).
 
 init({M,F,A}) ->
-    {ok, {{simple_one_for_one_terminate, 10, 10},
-          [{tcp_client, {M,F,A},
-            temporary, infinity, supervisor, [M]}]}}.
+    {ok, {{simple_one_for_one_terminate, 0, 1},
+          [{client, {M,F,A}, temporary, infinity, supervisor, [M]}]}}.
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 4228ff7fd3..746bb66eb5 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -20,6 +20,7 @@
 -export([start/0, stop/0, action/5, diagnostics/1]).
 
 -define(RPC_TIMEOUT, infinity).
+-define(WAIT_FOR_VM_ATTEMPTS, 5).
 
 -define(QUIET_OPT, "-q").
 -define(NODE_OPT, "-n").
@@ -44,22 +45,18 @@
 
 start() ->
     {ok, [[NodeStr|_]|_]} = init:get_argument(nodename),
-    FullCommand = init:get_plain_arguments(),
-    case FullCommand of
-        [] -> usage();
-        _ -> ok
-    end,
     {[Command0 | Args], Opts} =
-        rabbit_misc:get_options(
-          [{flag, ?QUIET_OPT}, {option, ?NODE_OPT, NodeStr},
-           {option, ?VHOST_OPT, "/"}],
-          FullCommand),
-    Opts1 = lists:map(fun({K, V}) ->
-                              case K of
-                                  ?NODE_OPT -> {?NODE_OPT, rabbit_misc:makenode(V)};
-                                  _    -> {K, V}
-                              end
-                      end, Opts),
+        case rabbit_misc:get_options([{flag, ?QUIET_OPT},
+                                      {option, ?NODE_OPT, NodeStr},
+                                      {option, ?VHOST_OPT, "/"}],
+                                     init:get_plain_arguments()) of
+            {[], _Opts}    -> usage();
+            CmdArgsAndOpts -> CmdArgsAndOpts
+        end,
+    Opts1 = [case K of
+                 ?NODE_OPT -> {?NODE_OPT, rabbit_misc:makenode(V)};
+                 _         -> {K, V}
+             end || {K, V} <- Opts],
     Command = list_to_atom(Command0),
     Quiet = proplists:get_bool(?QUIET_OPT, Opts1),
     Node = proplists:get_value(?NODE_OPT, Opts1),
@@ -77,24 +74,24 @@ start() ->
                 true  -> ok;
                 false -> io:format("...done.~n")
             end,
-            halt();
+            quit(0);
         {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
             print_error("invalid command '~s'",
                         [string:join([atom_to_list(Command) | Args], " ")]),
             usage();
         {error, Reason} ->
             print_error("~p", [Reason]),
-            halt(2);
+            quit(2);
         {badrpc, {'EXIT', Reason}} ->
             print_error("~p", [Reason]),
-            halt(2);
+            quit(2);
         {badrpc, Reason} ->
             print_error("unable to connect to node ~w: ~w", [Node, Reason]),
             print_badrpc_diagnostics(Node),
-            halt(2);
+            quit(2);
         Other ->
             print_error("~p", [Other]),
-            halt(2)
+            quit(2)
     end.
 
 fmt_stderr(Format, Args) -> rabbit_misc:format_stderr(Format ++ "~n", Args).
@@ -130,7 +127,7 @@ stop() ->
 
 usage() ->
     io:format("~s", [rabbit_ctl_usage:usage()]),
-    halt(1).
+    quit(1).
 
 action(stop, Node, [], _Opts, Inform) ->
     Inform("Stopping and halting node ~p", [Node]),
@@ -297,7 +294,30 @@ action(list_permissions, Node, [], Opts, Inform) ->
     VHost = proplists:get_value(?VHOST_OPT, Opts),
     Inform("Listing permissions in vhost ~p", [VHost]),
     display_list(call(Node, {rabbit_auth_backend_internal,
-                             list_vhost_permissions, [VHost]})).
+                             list_vhost_permissions, [VHost]}));
+
+action(wait, Node, [], _Opts, Inform) ->
+    Inform("Waiting for ~p", [Node]),
+    wait_for_application(Node, ?WAIT_FOR_VM_ATTEMPTS).
+
+wait_for_application(Node, Attempts) ->
+    case rpc_call(Node, application, which_applications, [infinity]) of
+        {badrpc, _} = E -> NewAttempts = Attempts - 1,
+                           case NewAttempts of
+                               0 -> E;
+                               _ -> wait_for_application0(Node, NewAttempts)
+                           end;
+        Apps            -> case proplists:is_defined(rabbit, Apps) of
+                               %% We've seen the node up; if it goes down
+                               %% die immediately.
+                               true  -> ok;
+                               false -> wait_for_application0(Node, 0)
+                           end
+    end.
+
+wait_for_application0(Node, Attempts) ->
+    timer:sleep(1000),
+    wait_for_application(Node, Attempts).
 
 default_if_empty(List, Default) when is_list(List) ->
     if List == [] ->
@@ -327,11 +347,11 @@ format_info_item(#resource{name = Name}) ->
     escape(Name);
 format_info_item({N1, N2, N3, N4} = Value) when
       ?IS_U8(N1), ?IS_U8(N2), ?IS_U8(N3), ?IS_U8(N4) ->
-    inet_parse:ntoa(Value);
+    rabbit_misc:ntoa(Value);
 format_info_item({K1, K2, K3, K4, K5, K6, K7, K8} = Value) when
       ?IS_U16(K1), ?IS_U16(K2), ?IS_U16(K3), ?IS_U16(K4),
       ?IS_U16(K5), ?IS_U16(K6), ?IS_U16(K7), ?IS_U16(K8) ->
-    inet_parse:ntoa(Value);
+    rabbit_misc:ntoa(Value);
 format_info_item(Value) when is_pid(Value) ->
     rabbit_misc:pid_to_string(Value);
 format_info_item(Value) when is_binary(Value) ->
@@ -393,3 +413,12 @@ prettify_typed_amqp_value(Type, Value) ->
         array   -> [prettify_typed_amqp_value(T, V) || {T, V} <- Value];
         _       -> Value
     end.
+
+% the slower shutdown on windows required to flush stdout
+quit(Status) ->
+    case os:type() of
+        {unix, _} ->
+            halt(Status);
+        {win32, _} ->
+            init:stop(Status)
+    end.
diff --git a/src/rabbit_direct.erl b/src/rabbit_direct.erl
new file mode 100644
index 0000000000..586563f61e
--- /dev/null
+++ b/src/rabbit_direct.erl
@@ -0,0 +1,79 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (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.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and
+%% limitations under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is VMware, Inc.
+%% Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
+%%
+
+-module(rabbit_direct).
+
+-export([boot/0, connect/4, start_channel/7]).
+
+-include("rabbit.hrl").
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(boot/0 :: () -> 'ok').
+-spec(connect/4 :: (binary(), binary(), binary(), rabbit_types:protocol()) ->
+                       {'ok', {rabbit_types:user(),
+                               rabbit_framing:amqp_table()}}).
+-spec(start_channel/7 ::
+        (rabbit_channel:channel_number(), pid(), rabbit_types:protocol(),
+         rabbit_types:user(), rabbit_types:vhost(), rabbit_framing:amqp_table(),
+         pid()) -> {'ok', pid()}).
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+boot() ->
+    {ok, _} =
+        supervisor2:start_child(
+            rabbit_sup,
+            {rabbit_direct_client_sup,
+             {rabbit_client_sup, start_link,
+              [{local, rabbit_direct_client_sup},
+               {rabbit_channel_sup, start_link, []}]},
+             transient, infinity, supervisor, [rabbit_client_sup]}),
+    ok.
+
+%%----------------------------------------------------------------------------
+
+connect(Username, Password, VHost, Protocol) ->
+    case lists:keymember(rabbit, 1, application:which_applications()) of
+        true  ->
+            try rabbit_access_control:user_pass_login(Username, Password) of
+                #user{} = User ->
+                    try rabbit_access_control:check_vhost_access(User, VHost) of
+                        ok -> {ok, {User,
+                                    rabbit_reader:server_properties(Protocol)}}
+                    catch
+                        exit:#amqp_error{name = access_refused} ->
+                            {error, access_refused}
+                    end
+            catch
+                exit:#amqp_error{name = access_refused} -> {error, auth_failure}
+            end;
+        false ->
+            {error, broker_not_found_on_node}
+    end.
+
+start_channel(Number, ClientChannelPid, Protocol, User, VHost, Capabilities,
+              Collector) ->
+    {ok, _, {ChannelPid, _}} =
+        supervisor2:start_child(
+            rabbit_direct_client_sup,
+            [{direct, Number, ClientChannelPid, Protocol, User, VHost,
+              Capabilities, Collector}]),
+    {ok, ChannelPid}.
diff --git a/src/rabbit_limiter.erl b/src/rabbit_limiter.erl
index 86ea7282d9..1b72dd761a 100644
--- a/src/rabbit_limiter.erl
+++ b/src/rabbit_limiter.erl
@@ -65,7 +65,7 @@ start_link(ChPid, UnackedMsgCount) ->
 limit(undefined, 0) ->
     ok;
 limit(LimiterPid, PrefetchCount) ->
-    gen_server2:call(LimiterPid, {limit, PrefetchCount}).
+    gen_server2:call(LimiterPid, {limit, PrefetchCount}, infinity).
 
 %% Ask the limiter whether the queue can deliver a message without
 %% breaching a limit
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 03317d7050..abc27c5f7d 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -55,6 +55,8 @@
 -export([now_ms/0]).
 -export([lock_file/1]).
 -export([const_ok/1, const/1]).
+-export([ntoa/1, ntoab/1]).
+-export([is_process_alive/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -191,6 +193,9 @@
 -spec(lock_file/1 :: (file:filename()) -> rabbit_types:ok_or_error('eexist')).
 -spec(const_ok/1 :: (any()) -> 'ok').
 -spec(const/1 :: (A) -> const(A)).
+-spec(ntoa/1 :: (inet:ip_address()) -> string()).
+-spec(ntoab/1 :: (inet:ip_address()) -> string()).
+-spec(is_process_alive/1 :: (pid()) -> boolean()).
 
 -endif.
 
@@ -237,11 +242,20 @@ assert_args_equivalence1(Orig, New, Name, Key) ->
         {Same, Same}  -> ok;
         {Orig1, New1} -> protocol_error(
                            precondition_failed,
-                           "inequivalent arg '~s' for ~s:  "
-                           "required ~w, received ~w",
-                           [Key, rabbit_misc:rs(Name), New1, Orig1])
+                           "inequivalent arg '~s' for ~s: "
+                           "received ~s but current is ~s",
+                           [Key, rs(Name), val(New1), val(Orig1)])
     end.
 
+val(undefined) ->
+    "none";
+val({Type, Value}) ->
+    Fmt = case is_binary(Value) of
+              true  -> "the value '~s' of type '~s'";
+              false -> "the value '~w' of type '~s'"
+          end,
+    lists:flatten(io_lib:format(Fmt, [Value, Type])).
+
 dirty_read(ReadSpec) ->
     case mnesia:dirty_read(ReadSpec) of
         [Result] -> {ok, Result};
@@ -338,8 +352,11 @@ throw_on_error(E, Thunk) ->
 with_exit_handler(Handler, Thunk) ->
     try
         Thunk()
-    catch exit:{R, _} when R =:= noproc; R =:= nodedown;
-                           R =:= normal; R =:= shutdown ->
+    catch
+        exit:{R, _} when R =:= noproc; R =:= nodedown;
+                         R =:= normal; R =:= shutdown ->
+            Handler();
+        exit:{{R, _}, _} when R =:= nodedown; R =:= shutdown ->
             Handler()
     end.
 
@@ -832,3 +849,26 @@ lock_file(Path) ->
 
 const_ok(_) -> ok.
 const(X) -> fun (_) -> X end.
+
+%% Format IPv4-mapped IPv6 addresses as IPv4, since they're what we see
+%% when IPv6 is enabled but not used (i.e. 99% of the time).
+ntoa({0,0,0,0,0,16#ffff,AB,CD}) ->
+    inet_parse:ntoa({AB bsr 8, AB rem 256, CD bsr 8, CD rem 256});
+ntoa(IP) ->
+    inet_parse:ntoa(IP).
+
+ntoab(IP) ->
+    Str = ntoa(IP),
+    case string:str(Str, ":") of
+        0 -> Str;
+        _ -> "[" ++ Str ++ "]"
+    end.
+
+is_process_alive(Pid) when node(Pid) =:= node() ->
+    erlang:is_process_alive(Pid);
+is_process_alive(Pid) ->
+    case rpc:call(node(Pid), erlang, is_process_alive, [Pid]) of
+        true -> true;
+        _    -> false
+    end.
+
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index a9b4e17745..a30f7996a4 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -371,15 +371,14 @@ init_db(ClusterNodes, Force) ->
                     %% True single disc node, attempt upgrade
                     ok = wait_for_tables(),
                     case rabbit_upgrade:maybe_upgrade() of
-                        ok                    -> ensure_schema_ok();
+                        ok                    -> ensure_schema_integrity();
                         version_not_available -> schema_ok_or_move()
                     end;
                 {[], true, _} ->
                     %% "Master" (i.e. without config) disc node in cluster,
                     %% verify schema
-                    ok = wait_for_tables(),
                     ensure_version_ok(rabbit_upgrade:read_version()),
-                    ensure_schema_ok();
+                    ensure_schema_integrity();
                 {[], false, _} ->
                     %% Nothing there at all, start from scratch
                     ok = create_schema();
@@ -396,7 +395,7 @@ init_db(ClusterNodes, Force) ->
                                                        true  -> disc;
                                                        false -> ram
                                                    end),
-                    ensure_schema_ok()
+                    ensure_schema_integrity()
             end;
         {error, Reason} ->
             %% one reason we may end up here is if we try to join
@@ -429,12 +428,6 @@ ensure_version_ok({ok, DiscVersion}) ->
 ensure_version_ok({error, _}) ->
     ok = rabbit_upgrade:write_version().
 
-ensure_schema_ok() ->
-    case check_schema_integrity() of
-        ok              -> ok;
-        {error, Reason} -> throw({error, {schema_invalid, Reason}})
-    end.
-
 create_schema() ->
     mnesia:stop(),
     rabbit_misc:ensure_ok(mnesia:create_schema([node()]),
@@ -443,7 +436,6 @@ create_schema() ->
                           cannot_start_mnesia),
     ok = create_tables(),
     ok = ensure_schema_integrity(),
-    ok = wait_for_tables(),
     ok = rabbit_upgrade:write_version().
 
 move_db() ->
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index 529e3e0706..7f3cf35faa 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -33,7 +33,7 @@
 
 -include("rabbit_msg_store.hrl").
 
--define(SYNC_INTERVAL,  5).   %% milliseconds
+-define(SYNC_INTERVAL,  25).   %% milliseconds
 -define(CLEAN_FILENAME, "clean.dot").
 -define(FILE_SUMMARY_FILENAME, "file_summary.ets").
 
@@ -738,45 +738,36 @@ handle_call({contains, Guid}, From, State) ->
 handle_cast({client_dying, CRef},
             State = #msstate { dying_clients = DyingClients }) ->
     DyingClients1 = sets:add_element(CRef, DyingClients),
-    write_message(CRef, <<>>, State #msstate { dying_clients = DyingClients1 });
+    noreply(write_message(CRef, <<>>,
+                          State #msstate { dying_clients = DyingClients1 }));
 
 handle_cast({client_delete, CRef}, State = #msstate { clients = Clients }) ->
     State1 = State #msstate { clients = dict:erase(CRef, Clients) },
     noreply(remove_message(CRef, CRef, clear_client(CRef, State1)));
 
 handle_cast({write, CRef, Guid},
-            State = #msstate { file_summary_ets   = FileSummaryEts,
-                               cur_file_cache_ets = CurFileCacheEts }) ->
+            State = #msstate { cur_file_cache_ets = CurFileCacheEts }) ->
     true = 0 =< ets:update_counter(CurFileCacheEts, Guid, {3, -1}),
     [{Guid, Msg, _CacheRefCount}] = ets:lookup(CurFileCacheEts, Guid),
-    case should_mask_action(CRef, Guid, State) of
-        {true, _Location} ->
-            noreply(State);
-        {false, not_found} ->
-            write_message(CRef, Guid, Msg, State);
-        {Mask, #msg_location { ref_count = 0, file = File,
-                               total_size = TotalSize }} ->
-            case {Mask, ets:lookup(FileSummaryEts, File)} of
-                {false, [#file_summary { locked = true }]} ->
-                    ok = index_delete(Guid, State),
-                    write_message(CRef, Guid, Msg, State);
-                {false_if_increment, [#file_summary { locked = true }]} ->
-                    %% The msg for Guid is older than the client death
-                    %% message, but as it is being GC'd currently,
-                    %% we'll have to write a new copy, which will then
-                    %% be younger, so ignore this write.
-                    noreply(State);
-                {_Mask, [#file_summary {}]} ->
-                    ok = index_update_ref_count(Guid, 1, State),
-                    State1 = client_confirm_if_on_disk(CRef, Guid, File, State),
-                    noreply(adjust_valid_total_size(File, TotalSize, State1))
-            end;
-        {_Mask, #msg_location { ref_count = RefCount, file = File }} ->
-            %% We already know about it, just update counter. Only
-            %% update field otherwise bad interaction with concurrent GC
-            ok = index_update_ref_count(Guid, RefCount + 1, State),
-            noreply(client_confirm_if_on_disk(CRef, Guid, File, State))
-    end;
+    noreply(
+      case write_action(should_mask_action(CRef, Guid, State), Guid, State) of
+          {write, State1} ->
+              write_message(CRef, Guid, Msg, State1);
+          {ignore, CurFile, State1 = #msstate { current_file = CurFile }} ->
+              State1;
+          {ignore, _File, State1} ->
+              true = ets:delete_object(CurFileCacheEts, {Guid, Msg, 0}),
+              State1;
+          {confirm, CurFile, State1 = #msstate { current_file = CurFile }}->
+              record_pending_confirm(CRef, Guid, State1);
+          {confirm, _File, State1} ->
+              true = ets:delete_object(CurFileCacheEts, {Guid, Msg, 0}),
+              update_pending_confirms(
+                fun (MsgOnDiskFun, CTG) ->
+                        MsgOnDiskFun(gb_sets:singleton(Guid), written),
+                        CTG
+                end, CRef, State1)
+      end);
 
 handle_cast({remove, CRef, Guids}, State) ->
     State1 = lists:foldl(
@@ -924,6 +915,37 @@ internal_sync(State = #msstate { current_file_handle = CurHdl,
     [client_confirm(CRef, Guids, written, State1) || {CRef, Guids} <- CGs],
     State1 #msstate { cref_to_guids = dict:new(), on_sync = [] }.
 
+write_action({true, not_found}, _Guid, State) ->
+    {ignore, undefined, State};
+write_action({true, #msg_location { file = File }}, _Guid, State) ->
+    {ignore, File, State};
+write_action({false, not_found}, _Guid, State) ->
+    {write, State};
+write_action({Mask, #msg_location { ref_count = 0, file = File,
+                                    total_size = TotalSize }},
+             Guid, State = #msstate { file_summary_ets = FileSummaryEts }) ->
+    case {Mask, ets:lookup(FileSummaryEts, File)} of
+        {false, [#file_summary { locked = true }]} ->
+            ok = index_delete(Guid, State),
+            {write, State};
+        {false_if_increment, [#file_summary { locked = true }]} ->
+            %% The msg for Guid is older than the client death
+            %% message, but as it is being GC'd currently we'll have
+            %% to write a new copy, which will then be younger, so
+            %% ignore this write.
+            {ignore, File, State};
+        {_Mask, [#file_summary {}]} ->
+            ok = index_update_ref_count(Guid, 1, State),
+            State1 = adjust_valid_total_size(File, TotalSize, State),
+            {confirm, File, State1}
+    end;
+write_action({_Mask, #msg_location { ref_count = RefCount, file = File }},
+             Guid, State) ->
+    ok = index_update_ref_count(Guid, RefCount + 1, State),
+    %% We already know about it, just update counter. Only update
+    %% field otherwise bad interaction with concurrent GC
+    {confirm, File, State}.
+
 write_message(CRef, Guid, Msg, State) ->
     write_message(Guid, Msg, record_pending_confirm(CRef, Guid, State)).
 
@@ -943,11 +965,10 @@ write_message(Guid, Msg,
     [_,_] = ets:update_counter(FileSummaryEts, CurFile,
                                [{#file_summary.valid_total_size, TotalSize},
                                 {#file_summary.file_size,        TotalSize}]),
-    NextOffset = CurOffset + TotalSize,
-    noreply(maybe_roll_to_new_file(
-              NextOffset, State #msstate {
-                            sum_valid_data = SumValid    + TotalSize,
-                            sum_file_size  = SumFileSize + TotalSize })).
+    maybe_roll_to_new_file(CurOffset + TotalSize,
+                           State #msstate {
+                             sum_valid_data = SumValid    + TotalSize,
+                             sum_file_size  = SumFileSize + TotalSize }).
 
 read_message(Guid, From,
              State = #msstate { dedup_cache_ets = DedupCacheEts }) ->
@@ -1134,16 +1155,6 @@ record_pending_confirm(CRef, Guid, State) ->
                           gb_sets:singleton(Guid), CTG)
       end, CRef, State).
 
-client_confirm_if_on_disk(CRef, Guid, CurFile,
-                          State = #msstate { current_file = CurFile }) ->
-    record_pending_confirm(CRef, Guid, State);
-client_confirm_if_on_disk(CRef, Guid, _File, State) ->
-    update_pending_confirms(
-      fun (MsgOnDiskFun, CTG) ->
-              MsgOnDiskFun(gb_sets:singleton(Guid), written),
-              CTG
-      end, CRef, State).
-
 client_confirm(CRef, Guids, ActionTaken, State) ->
     update_pending_confirms(
       fun (MsgOnDiskFun, CTG) ->
diff --git a/src/rabbit_multi.erl b/src/rabbit_multi.erl
deleted file mode 100644
index 7c07c4fedb..0000000000
--- a/src/rabbit_multi.erl
+++ /dev/null
@@ -1,347 +0,0 @@
-%% The contents of this file are subject to the Mozilla Public License
-%% Version 1.1 (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.mozilla.org/MPL/
-%%
-%% Software distributed under the License is distributed on an "AS IS"
-%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
-%% the License for the specific language governing rights and
-%% limitations under the License.
-%%
-%% The Original Code is RabbitMQ.
-%%
-%% The Initial Developer of the Original Code is VMware, Inc.
-%% Copyright (c) 2007-2011 VMware, Inc.  All rights reserved.
-%%
-
--module(rabbit_multi).
--include("rabbit.hrl").
-
--export([start/0, stop/0]).
-
--define(RPC_SLEEP, 500).
-
-%%----------------------------------------------------------------------------
-
--ifdef(use_specs).
-
--spec(start/0 :: () -> no_return()).
--spec(stop/0 :: () -> 'ok').
--spec(usage/0 :: () -> no_return()).
-
--endif.
-
-%%----------------------------------------------------------------------------
-
-start() ->
-    RpcTimeout =
-        case init:get_argument(maxwait) of
-            {ok,[[N1]]} -> 1000 * list_to_integer(N1);
-            _           -> ?MAX_WAIT
-        end,
-    case init:get_plain_arguments() of
-        [] ->
-            usage();
-        FullCommand ->
-            {Command, Args} = parse_args(FullCommand),
-            case catch action(Command, Args, RpcTimeout) of
-                ok ->
-                    io:format("done.~n"),
-                    halt();
-                {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
-                    print_error("invalid command '~s'",
-                                [string:join(FullCommand, " ")]),
-                    usage();
-                timeout ->
-                    print_error("timeout starting some nodes.", []),
-                    halt(1);
-                Other ->
-                    print_error("~p", [Other]),
-                    halt(2)
-            end
-    end.
-
-print_error(Format, Args) ->
-    rabbit_misc:format_stderr("Error: " ++ Format ++ "~n", Args).
-
-parse_args([Command | Args]) ->
-    {list_to_atom(Command), Args}.
-
-stop() ->
-    ok.
-
-usage() ->
-    io:format("~s", [rabbit_multi_usage:usage()]),
-    halt(1).
-
-action(start_all, [NodeCount], RpcTimeout) ->
-    io:format("Starting all nodes...~n", []),
-    application:load(rabbit),
-    {_NodeNamePrefix, NodeHost} = NodeName = rabbit_misc:nodeparts(
-                                               getenv("RABBITMQ_NODENAME")),
-    case net_adm:names(NodeHost) of
-        {error, EpmdReason} ->
-            throw({cannot_connect_to_epmd, NodeHost, EpmdReason});
-        {ok, _} ->
-            ok
-    end,
-    {NodePids, Running} =
-        case list_to_integer(NodeCount) of
-            1 -> {NodePid, Started} = start_node(rabbit_misc:makenode(NodeName),
-                                                 RpcTimeout),
-                 {[NodePid], Started};
-            N -> start_nodes(N, N, [], true, NodeName,
-                             get_node_tcp_listener(), RpcTimeout)
-        end,
-    write_pids_file(NodePids),
-    case Running of
-        true  -> ok;
-        false -> timeout
-    end;
-
-action(status, [], RpcTimeout) ->
-    io:format("Status of all running nodes...~n", []),
-    call_all_nodes(
-      fun ({Node, Pid}) ->
-              RabbitRunning =
-                  case is_rabbit_running(Node, RpcTimeout) of
-                      false -> not_running;
-                      true  -> running
-                  end,
-              io:format("Node '~p' with Pid ~p: ~p~n",
-                        [Node, Pid, RabbitRunning])
-      end);
-
-action(stop_all, [], RpcTimeout) ->
-    io:format("Stopping all nodes...~n", []),
-    call_all_nodes(fun ({Node, Pid}) ->
-                           io:format("Stopping node ~p~n", [Node]),
-                           rpc:call(Node, rabbit, stop_and_halt, []),
-                           case kill_wait(Pid, RpcTimeout, false) of
-                               false -> kill_wait(Pid, RpcTimeout, true);
-                               true  -> ok
-                           end,
-                           io:format("OK~n", [])
-                   end),
-    delete_pids_file();
-
-action(rotate_logs, [], RpcTimeout) ->
-    action(rotate_logs, [""], RpcTimeout);
-
-action(rotate_logs, [Suffix], RpcTimeout) ->
-    io:format("Rotating logs for all nodes...~n", []),
-    BinarySuffix = list_to_binary(Suffix),
-    call_all_nodes(
-      fun ({Node, _}) ->
-              io:format("Rotating logs for node ~p", [Node]),
-              case rpc:call(Node, rabbit, rotate_logs,
-                            [BinarySuffix], RpcTimeout) of
-                  {badrpc, Error} -> io:format(": ~p.~n", [Error]);
-                  ok              -> io:format(": ok.~n", [])
-              end
-      end).
-
-%% PNodePid is the list of PIDs
-%% Running is a boolean exhibiting success at some moment
-start_nodes(0, _, PNodePid, Running, _, _, _) -> {PNodePid, Running};
-
-start_nodes(N, Total, PNodePid, Running, NodeNameBase, Listener, RpcTimeout) ->
-    {NodePre, NodeSuff} = NodeNameBase,
-    NodeNumber = Total - N,
-    NodePre1 = case NodeNumber of
-                   %% For compatibility with running a single node
-                   0 -> NodePre;
-                   _ -> NodePre ++ "_" ++ integer_to_list(NodeNumber)
-               end,
-    Node = rabbit_misc:makenode({NodePre1, NodeSuff}),
-    os:putenv("RABBITMQ_NODENAME", atom_to_list(Node)),
-    case Listener of
-        {NodeIpAddress, NodePortBase} ->
-            NodePort = NodePortBase + NodeNumber,
-            os:putenv("RABBITMQ_NODE_PORT", integer_to_list(NodePort)),
-            os:putenv("RABBITMQ_NODE_IP_ADDRESS", NodeIpAddress);
-        undefined ->
-            ok
-    end,
-    {NodePid, Started} = start_node(Node, RpcTimeout),
-    start_nodes(N - 1, Total, [NodePid | PNodePid],
-                Started and Running, NodeNameBase, Listener, RpcTimeout).
-
-start_node(Node, RpcTimeout) ->
-    io:format("Starting node ~s...~n", [Node]),
-    case rpc:call(Node, os, getpid, []) of
-        {badrpc, _} ->
-            Port = run_rabbitmq_server(),
-            Started = wait_for_rabbit_to_start(Node, RpcTimeout, Port),
-            Pid = case rpc:call(Node, os, getpid, []) of
-                      {badrpc, _} -> throw(cannot_get_pid);
-                      PidS -> list_to_integer(PidS)
-                  end,
-            io:format("~s~n", [case Started of
-                                   true  -> "OK";
-                                   false -> "timeout"
-                               end]),
-            {{Node, Pid}, Started};
-        PidS ->
-            Pid = list_to_integer(PidS),
-            throw({node_already_running, Node, Pid})
-    end.
-
-wait_for_rabbit_to_start(_ , RpcTimeout, _) when RpcTimeout < 0 ->
-    false;
-wait_for_rabbit_to_start(Node, RpcTimeout, Port) ->
-    case is_rabbit_running(Node, RpcTimeout) of
-        true  -> true;
-        false -> receive
-                     {'EXIT', Port, PosixCode} ->
-                         throw({node_start_failed, PosixCode})
-                 after ?RPC_SLEEP ->
-                         wait_for_rabbit_to_start(
-                           Node, RpcTimeout - ?RPC_SLEEP, Port)
-                 end
-    end.
-
-run_rabbitmq_server() ->
-    with_os([{unix, fun run_rabbitmq_server_unix/0},
-             {win32, fun run_rabbitmq_server_win32/0}]).
-
-run_rabbitmq_server_unix() ->
-    CmdLine = getenv("RABBITMQ_SCRIPT_HOME") ++ "/rabbitmq-server -noinput",
-    erlang:open_port({spawn, CmdLine}, [nouse_stdio]).
-
-run_rabbitmq_server_win32() ->
-    Cmd = filename:nativename(os:find_executable("cmd")),
-    CmdLine = "\"" ++ getenv("RABBITMQ_SCRIPT_HOME") ++
-              "\\rabbitmq-server.bat\" -noinput -detached",
-    erlang:open_port({spawn_executable, Cmd},
-                     [{arg0, Cmd}, {args, ["/q", "/s", "/c", CmdLine]},
-                      nouse_stdio]).
-
-is_rabbit_running(Node, RpcTimeout) ->
-    case rpc:call(Node, rabbit, status, [], RpcTimeout) of
-        {badrpc, _} -> false;
-        Status      -> case proplists:get_value(running_applications, Status) of
-                           undefined -> false;
-                           Apps      -> lists:keymember(rabbit, 1, Apps)
-                       end
-    end.
-
-with_os(Handlers) ->
-    {OsFamily, _} = os:type(),
-    case proplists:get_value(OsFamily, Handlers) of
-        undefined -> throw({unsupported_os, OsFamily});
-        Handler   -> Handler()
-    end.
-
-pids_file() -> getenv("RABBITMQ_PIDS_FILE").
-
-write_pids_file(Pids) ->
-    FileName = pids_file(),
-    Handle = case file:open(FileName, [write]) of
-                 {ok, Device} ->
-                     Device;
-                 {error, Reason} ->
-                     throw({cannot_create_pids_file, FileName, Reason})
-             end,
-    try
-        ok = io:write(Handle, Pids),
-        ok = io:put_chars(Handle, [$.])
-    after
-        case file:close(Handle) of
-            ok -> ok;
-            {error, Reason1} ->
-                throw({cannot_create_pids_file, FileName, Reason1})
-        end
-    end,
-    ok.
-
-delete_pids_file() ->
-    FileName = pids_file(),
-    case file:delete(FileName) of
-        ok              -> ok;
-        {error, enoent} -> ok;
-        {error, Reason} -> throw({cannot_delete_pids_file, FileName, Reason})
-    end.
-
-read_pids_file() ->
-    FileName = pids_file(),
-    case file:consult(FileName) of
-        {ok, [Pids]}    -> Pids;
-        {error, enoent} -> [];
-        {error, Reason} -> throw({cannot_read_pids_file, FileName, Reason})
-    end.
-
-kill_wait(Pid, TimeLeft, Forceful) when TimeLeft < 0 ->
-    Cmd = with_os([{unix, fun () -> if Forceful -> "kill -9";
-                                       true     -> "kill"
-                                    end
-                          end},
-                   %% Kill forcefully always on Windows, since erl.exe
-                   %% seems to completely ignore non-forceful killing
-                   %% even when everything is working
-                   {win32, fun () -> "taskkill /f /pid" end}]),
-    os:cmd(Cmd ++ " " ++ integer_to_list(Pid)),
-    false; % Don't assume what we did just worked!
-
-% Returns true if the process is dead, false otherwise.
-kill_wait(Pid, TimeLeft, Forceful) ->
-    timer:sleep(?RPC_SLEEP),
-    io:format(".", []),
-    is_dead(Pid) orelse kill_wait(Pid, TimeLeft - ?RPC_SLEEP, Forceful).
-
-% Test using some OS clunkiness since we shouldn't trust
-% rpc:call(os, getpid, []) at this point
-is_dead(Pid) ->
-    PidS = integer_to_list(Pid),
-    with_os([{unix, fun () ->
-                            system("kill -0 " ++ PidS
-                                   ++ " >/dev/null 2>&1") /= 0
-                    end},
-             {win32, fun () ->
-                             Res = os:cmd("tasklist /nh /fi \"pid eq " ++
-                                          PidS ++ "\" 2>&1"),
-                             case re:run(Res, "erl\\.exe", [{capture, none}]) of
-                                 match -> false;
-                                 _     -> true
-                             end
-                     end}]).
-
-% Like system(3)
-system(Cmd) ->
-    ShCmd = "sh -c '" ++ escape_quotes(Cmd) ++ "'",
-    Port = erlang:open_port({spawn, ShCmd}, [exit_status,nouse_stdio]),
-    receive {Port, {exit_status, Status}} -> Status end.
-
-% Escape the quotes in a shell command so that it can be used in "sh -c 'cmd'"
-escape_quotes(Cmd) ->
-    lists:flatten(lists:map(fun ($') -> "'\\''"; (Ch) -> Ch end, Cmd)).
-
-call_all_nodes(Func) ->
-    case read_pids_file() of
-        []       -> throw(no_nodes_running);
-        NodePids -> lists:foreach(Func, NodePids)
-    end.
-
-getenv(Var) ->
-    case os:getenv(Var) of
-        false -> throw({missing_env_var, Var});
-        Value -> Value
-    end.
-
-get_node_tcp_listener() ->
-    try
-        {getenv("RABBITMQ_NODE_IP_ADDRESS"),
-         list_to_integer(getenv("RABBITMQ_NODE_PORT"))}
-    catch _ ->
-            case application:get_env(rabbit, tcp_listeners) of
-                {ok, [{_IpAddy, _Port} = Listener]} ->
-                    Listener;
-                {ok, []} ->
-                    undefined;
-                {ok, Other} ->
-                    throw({cannot_start_multiple_nodes, multiple_tcp_listeners,
-                           Other});
-                undefined ->
-                    throw({missing_configuration, tcp_listeners})
-            end
-    end.
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 9788c922bf..36f61628b8 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -16,15 +16,15 @@
 
 -module(rabbit_networking).
 
--export([boot/0, start/0, start_tcp_listener/2, start_ssl_listener/3,
-         stop_tcp_listener/2, on_node_down/1, active_listeners/0,
+-export([boot/0, start/0, start_tcp_listener/1, start_ssl_listener/2,
+         stop_tcp_listener/1, on_node_down/1, active_listeners/0,
          node_listeners/1, connections/0, connection_info_keys/0,
          connection_info/1, connection_info/2,
          connection_info_all/0, connection_info_all/1,
          close_connection/2]).
 
 %%used by TCP-based transports, e.g. STOMP adapter
--export([check_tcp_listener_address/3]).
+-export([check_tcp_listener_address/2]).
 
 -export([tcp_listener_started/3, tcp_listener_stopped/3,
          start_client/1, start_ssl_client/2]).
@@ -32,29 +32,26 @@
 -include("rabbit.hrl").
 -include_lib("kernel/include/inet.hrl").
 
--define(RABBIT_TCP_OPTS, [
-        binary,
-        {packet, raw}, % no packaging
-        {reuseaddr, true}, % allow rebind without waiting
-        {backlog, 128}, % use the maximum listen(2) backlog value
-        %% {nodelay, true}, % TCP_NODELAY - disable Nagle's alg.
-        %% {delay_send, true},
-        {exit_on_close, false}
-    ]).
-
 -define(SSL_TIMEOUT, 5). %% seconds
 
+-define(FIRST_TEST_BIND_PORT, 10000).
+
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
 
 -export_type([ip_port/0, hostname/0]).
 
+-type(family() :: atom()).
+-type(listener_config() :: ip_port() |
+                           {hostname(), ip_port()} |
+                           {hostname(), ip_port(), family()}).
+
 -spec(start/0 :: () -> 'ok').
--spec(start_tcp_listener/2 :: (hostname(), ip_port()) -> 'ok').
--spec(start_ssl_listener/3 :: (hostname(), ip_port(), rabbit_types:infos())
-                              -> 'ok').
--spec(stop_tcp_listener/2 :: (hostname(), ip_port()) -> 'ok').
+-spec(start_tcp_listener/1 :: (listener_config()) -> 'ok').
+-spec(start_ssl_listener/2 ::
+        (listener_config(), rabbit_types:infos()) -> 'ok').
+-spec(stop_tcp_listener/1 :: (listener_config()) -> 'ok').
 -spec(active_listeners/0 :: () -> [rabbit_types:listener()]).
 -spec(node_listeners/1 :: (node()) -> [rabbit_types:listener()]).
 -spec(connections/0 :: () -> [rabbit_types:connection()]).
@@ -69,8 +66,8 @@
         (rabbit_types:info_keys()) -> [rabbit_types:infos()]).
 -spec(close_connection/2 :: (pid(), string()) -> 'ok').
 -spec(on_node_down/1 :: (node()) -> 'ok').
--spec(check_tcp_listener_address/3 ::
-        (atom(), hostname(), ip_port()) -> {inet:ip_address(), atom()}).
+-spec(check_tcp_listener_address/2 :: (atom(), listener_config())
+        -> [{inet:ip_address(), ip_port(), family(), atom()}]).
 
 -endif.
 
@@ -83,7 +80,7 @@ boot() ->
 
 boot_tcp() ->
     {ok, TcpListeners} = application:get_env(tcp_listeners),
-    [ok = start_tcp_listener(Host, Port) || {Host, Port} <- TcpListeners],
+    [ok = start_tcp_listener(Listener) || Listener <- TcpListeners],
     ok.
 
 boot_ssl() ->
@@ -103,75 +100,111 @@ boot_ssl() ->
                                                  end}
                                    | SslOptsConfig]
                 end,
-            [start_ssl_listener(Host, Port, SslOpts) || {Host, Port} <- SslListeners],
+            [start_ssl_listener(Listener, SslOpts) || Listener <- SslListeners],
             ok
     end.
 
 start() ->
-    {ok,_} = supervisor:start_child(
+    {ok,_} = supervisor2:start_child(
                rabbit_sup,
                {rabbit_tcp_client_sup,
-                {tcp_client_sup, start_link,
+                {rabbit_client_sup, start_link,
                  [{local, rabbit_tcp_client_sup},
                   {rabbit_connection_sup,start_link,[]}]},
-                transient, infinity, supervisor, [tcp_client_sup]}),
+                transient, infinity, supervisor, [rabbit_client_sup]}),
     ok.
 
-getaddr(Host) ->
-    %% inet_parse:address takes care of ip string, like "0.0.0.0"
-    %% inet:getaddr returns immediately for ip tuple {0,0,0,0},
-    %%  and runs 'inet_gethost' port process for dns lookups.
-    %% On Windows inet:getaddr runs dns resolver for ip string, which may fail.
+%% inet_parse:address takes care of ip string, like "0.0.0.0"
+%% inet:getaddr returns immediately for ip tuple {0,0,0,0},
+%%  and runs 'inet_gethost' port process for dns lookups.
+%% On Windows inet:getaddr runs dns resolver for ip string, which may fail.
+
+getaddr(Host, Family) ->
     case inet_parse:address(Host) of
-        {ok, IPAddress1} -> IPAddress1;
-        {error, _} ->
-            case inet:getaddr(Host, inet) of
-                {ok, IPAddress2} -> IPAddress2;
-                {error, Reason} ->
-                    error_logger:error_msg("invalid host ~p - ~p~n",
-                                           [Host, Reason]),
-                    throw({error, {invalid_host, Host, Reason}})
-            end
+        {ok, IPAddress} -> [{IPAddress, resolve_family(IPAddress, Family)}];
+        {error, _}      -> gethostaddr(Host, Family)
     end.
 
-check_tcp_listener_address(NamePrefix, Host, Port) ->
-    IPAddress = getaddr(Host),
+gethostaddr(Host, auto) ->
+    Lookups = [{Family, inet:getaddr(Host, Family)} || Family <- [inet, inet6]],
+    case [{IP, Family} || {Family, {ok, IP}} <- Lookups] of
+        []  -> host_lookup_error(Host, Lookups);
+        IPs -> IPs
+    end;
+
+gethostaddr(Host, Family) ->
+    case inet:getaddr(Host, Family) of
+        {ok, IPAddress} -> [{IPAddress, Family}];
+        {error, Reason} -> host_lookup_error(Host, Reason)
+    end.
+
+host_lookup_error(Host, Reason) ->
+    error_logger:error_msg("invalid host ~p - ~p~n", [Host, Reason]),
+    throw({error, {invalid_host, Host, Reason}}).
+
+resolve_family({_,_,_,_},         auto) -> inet;
+resolve_family({_,_,_,_,_,_,_,_}, auto) -> inet6;
+resolve_family(IP,                auto) -> throw({error, {strange_family, IP}});
+resolve_family(_,                 F)    -> F.
+
+check_tcp_listener_address(NamePrefix, Port) when is_integer(Port) ->
+    check_tcp_listener_address_auto(NamePrefix, Port);
+
+check_tcp_listener_address(NamePrefix, {"auto", Port}) ->
+    %% Variant to prevent lots of hacking around in bash and batch files
+    check_tcp_listener_address_auto(NamePrefix, Port);
+
+check_tcp_listener_address(NamePrefix, {Host, Port}) ->
+    %% auto: determine family IPv4 / IPv6 after converting to IP address
+    check_tcp_listener_address(NamePrefix, {Host, Port, auto});
+
+check_tcp_listener_address(NamePrefix, {Host, Port, Family0}) ->
     if is_integer(Port) andalso (Port >= 0) andalso (Port =< 65535) -> ok;
        true -> error_logger:error_msg("invalid port ~p - not 0..65535~n",
                                       [Port]),
                throw({error, {invalid_port, Port}})
     end,
-    Name = rabbit_misc:tcp_name(NamePrefix, IPAddress, Port),
-    {IPAddress, Name}.
+    [{IPAddress, Port, Family,
+      rabbit_misc:tcp_name(NamePrefix, IPAddress, Port)} ||
+        {IPAddress, Family} <- getaddr(Host, Family0)].
+
+check_tcp_listener_address_auto(NamePrefix, Port) ->
+    lists:append([check_tcp_listener_address(NamePrefix, Listener) ||
+                     Listener <- port_to_listeners(Port)]).
 
-start_tcp_listener(Host, Port) ->
-    start_listener(Host, Port, amqp, "TCP Listener",
+start_tcp_listener(Listener) ->
+    start_listener(Listener, amqp, "TCP Listener",
                    {?MODULE, start_client, []}).
 
-start_ssl_listener(Host, Port, SslOpts) ->
-    start_listener(Host, Port, 'amqp/ssl', "SSL Listener",
+start_ssl_listener(Listener, SslOpts) ->
+    start_listener(Listener, 'amqp/ssl', "SSL Listener",
                    {?MODULE, start_ssl_client, [SslOpts]}).
 
-start_listener(Host, Port, Protocol, Label, OnConnect) ->
-    {IPAddress, Name} =
-        check_tcp_listener_address(rabbit_tcp_listener_sup, Host, Port),
+start_listener(Listener, Protocol, Label, OnConnect) ->
+    [start_listener0(Spec, Protocol, Label, OnConnect) ||
+        Spec <- check_tcp_listener_address(rabbit_tcp_listener_sup, Listener)],
+    ok.
+
+start_listener0({IPAddress, Port, Family, Name}, Protocol, Label, OnConnect) ->
     {ok,_} = supervisor:start_child(
                rabbit_sup,
                {Name,
                 {tcp_listener_sup, start_link,
-                 [IPAddress, Port, ?RABBIT_TCP_OPTS ,
+                 [IPAddress, Port, [Family | tcp_opts()],
                   {?MODULE, tcp_listener_started, [Protocol]},
                   {?MODULE, tcp_listener_stopped, [Protocol]},
                   OnConnect, Label]},
-                transient, infinity, supervisor, [tcp_listener_sup]}),
+                transient, infinity, supervisor, [tcp_listener_sup]}).
+
+stop_tcp_listener(Listener) ->
+    [stop_tcp_listener0(Spec) ||
+        Spec <- check_tcp_listener_address(rabbit_tcp_listener_sup, Listener)],
     ok.
 
-stop_tcp_listener(Host, Port) ->
-    IPAddress = getaddr(Host),
+stop_tcp_listener0({IPAddress, Port, _Family, Name}) ->
     Name = rabbit_misc:tcp_name(rabbit_tcp_listener_sup, IPAddress, Port),
     ok = supervisor:terminate_child(rabbit_sup, Name),
-    ok = supervisor:delete_child(rabbit_sup, Name),
-    ok.
+    ok = supervisor:delete_child(rabbit_sup, Name).
 
 tcp_listener_started(Protocol, IPAddress, Port) ->
     %% We need the ip to distinguish e.g. 0.0.0.0 and 127.0.0.1
@@ -252,15 +285,106 @@ close_connection(Pid, Explanation) ->
 %%--------------------------------------------------------------------
 
 tcp_host({0,0,0,0}) ->
-    {ok, Hostname} = inet:gethostname(),
-    case inet:gethostbyname(Hostname) of
-        {ok, #hostent{h_name = Name}} -> Name;
-        {error, _Reason} -> Hostname
-    end;
+    hostname();
+
+tcp_host({0,0,0,0,0,0,0,0}) ->
+    hostname();
+
 tcp_host(IPAddress) ->
     case inet:gethostbyaddr(IPAddress) of
         {ok, #hostent{h_name = Name}} -> Name;
-        {error, _Reason} -> inet_parse:ntoa(IPAddress)
+        {error, _Reason} -> rabbit_misc:ntoa(IPAddress)
+    end.
+
+hostname() ->
+    {ok, Hostname} = inet:gethostname(),
+    case inet:gethostbyname(Hostname) of
+        {ok,    #hostent{h_name = Name}} -> Name;
+        {error, _Reason}                 -> Hostname
     end.
 
 cmap(F) -> rabbit_misc:filter_exit_map(F, connections()).
+
+tcp_opts() ->
+    {ok, Opts} = application:get_env(rabbit, tcp_listen_options),
+    Opts.
+
+%%--------------------------------------------------------------------
+
+%% There are three kinds of machine (for our purposes).
+%%
+%% * Those which treat IPv4 addresses as a special kind of IPv6 address
+%%   ("Single stack")
+%%   - Linux by default, Windows Vista and later
+%%   - We also treat any (hypothetical?) IPv6-only machine the same way
+%% * Those which consider IPv6 and IPv4 to be completely separate things
+%%   ("Dual stack")
+%%   - OpenBSD, Windows XP / 2003, Linux if so configured
+%% * Those which do not support IPv6.
+%%   - Ancient/weird OSes, Linux if so configured
+%%
+%% How to reconfigure Linux to test this:
+%% Single stack (default):
+%% echo 0 > /proc/sys/net/ipv6/bindv6only
+%% Dual stack:
+%% echo 1 > /proc/sys/net/ipv6/bindv6only
+%% IPv4 only:
+%% add ipv6.disable=1 to GRUB_CMDLINE_LINUX_DEFAULT in /etc/default/grub then
+%% sudo update-grub && sudo reboot
+%%
+%% This matters in (and only in) the case where the sysadmin (or the
+%% app descriptor) has only supplied a port and we wish to bind to
+%% "all addresses". This means different things depending on whether
+%% we're single or dual stack. On single stack binding to "::"
+%% implicitly includes all IPv4 addresses, and subsequently attempting
+%% to bind to "0.0.0.0" will fail. On dual stack, binding to "::" will
+%% only bind to IPv6 addresses, and we need another listener bound to
+%% "0.0.0.0" for IPv4. Finally, on IPv4-only systems we of course only
+%% want to bind to "0.0.0.0".
+%%
+%% Unfortunately it seems there is no way to detect single vs dual stack
+%% apart from attempting to bind to the port.
+port_to_listeners(Port) ->
+    IPv4 = {"0.0.0.0", Port, inet},
+    IPv6 = {"::",      Port, inet6},
+    case ipv6_status(?FIRST_TEST_BIND_PORT) of
+        single_stack -> [IPv6];
+        ipv6_only    -> [IPv6];
+        dual_stack   -> [IPv6, IPv4];
+        ipv4_only    -> [IPv4]
+    end.
+
+ipv6_status(TestPort) ->
+    IPv4 = [inet,  {ip, {0,0,0,0}}],
+    IPv6 = [inet6, {ip, {0,0,0,0,0,0,0,0}}],
+    case gen_tcp:listen(TestPort, IPv6) of
+        {ok, LSock6} ->
+            case gen_tcp:listen(TestPort, IPv4) of
+                {ok, LSock4} ->
+                    %% Dual stack
+                    gen_tcp:close(LSock6),
+                    gen_tcp:close(LSock4),
+                    dual_stack;
+                %% Checking the error here would only let us
+                %% distinguish single stack IPv6 / IPv4 vs IPv6 only,
+                %% which we figure out below anyway.
+                {error, _} ->
+                    gen_tcp:close(LSock6),
+                    case gen_tcp:listen(TestPort, IPv4) of
+                        %% Single stack
+                        {ok, LSock4}            -> gen_tcp:close(LSock4),
+                                                   single_stack;
+                        %% IPv6-only machine. Welcome to the future.
+                        {error, eafnosupport}   -> ipv6_only;
+                        %% Dual stack machine with something already
+                        %% on IPv4.
+                        {error, _}              -> ipv6_status(TestPort + 1)
+                    end
+            end;
+        {error, eafnosupport} ->
+            %% IPv4-only machine. Welcome to the 90s.
+            ipv4_only;
+        {error, _} ->
+            %% Port in use
+            ipv6_status(TestPort + 1)
+    end.
diff --git a/src/rabbit_node_monitor.erl b/src/rabbit_node_monitor.erl
index e4bc1cdc5a..817abaa2bd 100644
--- a/src/rabbit_node_monitor.erl
+++ b/src/rabbit_node_monitor.erl
@@ -22,14 +22,41 @@
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
+-export([notify_cluster/0, rabbit_running_on/1]).
 
 -define(SERVER, ?MODULE).
+-define(RABBIT_UP_RPC_TIMEOUT, 2000).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(rabbit_running_on/1 :: (node()) -> 'ok').
+-spec(notify_cluster/0 :: () -> 'ok').
+
+-endif.
 
 %%--------------------------------------------------------------------
 
 start_link() ->
     gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 
+rabbit_running_on(Node) ->
+    gen_server:cast(rabbit_node_monitor, {rabbit_running_on, Node}).
+
+notify_cluster() ->
+    Node = node(),
+    Nodes = rabbit_mnesia:running_clustered_nodes() -- [Node],
+    %% notify other rabbits of this rabbit
+    case rpc:multicall(Nodes, rabbit_node_monitor, rabbit_running_on,
+                       [Node], ?RABBIT_UP_RPC_TIMEOUT) of
+        {_, [] } -> ok;
+        {_, Bad} -> rabbit_log:info("failed to contact nodes ~p~n", [Bad])
+    end,
+    %% register other active rabbits with this rabbit
+    [ rabbit_node_monitor:rabbit_running_on(N) || N <- Nodes ],
+    ok.
+
 %%--------------------------------------------------------------------
 
 init([]) ->
@@ -39,19 +66,20 @@ init([]) ->
 handle_call(_Request, _From, State) ->
     {noreply, State}.
 
+handle_cast({rabbit_running_on, Node}, State) ->
+    rabbit_log:info("node ~p up~n", [Node]),
+    erlang:monitor(process, {rabbit, Node}),
+    {noreply, State};
 handle_cast(_Msg, State) ->
     {noreply, State}.
 
-handle_info({nodeup, Node}, State) ->
-    rabbit_log:info("node ~p up", [Node]),
-    {noreply, State};
 handle_info({nodedown, Node}, State) ->
-    rabbit_log:info("node ~p down", [Node]),
-    %% TODO: This may turn out to be a performance hog when there are
-    %% lots of nodes.  We really only need to execute this code on
-    %% *one* node, rather than all of them.
-    ok = rabbit_networking:on_node_down(Node),
-    ok = rabbit_amqqueue:on_node_down(Node),
+    rabbit_log:info("node ~p down~n", [Node]),
+    ok = handle_dead_rabbit(Node),
+     {noreply, State};
+handle_info({'DOWN', _MRef, process, {rabbit, Node}, _Reason}, State) ->
+    rabbit_log:info("node ~p lost 'rabbit'~n", [Node]),
+    ok = handle_dead_rabbit(Node),
     {noreply, State};
 handle_info(_Info, State) ->
     {noreply, State}.
@@ -64,3 +92,10 @@ code_change(_OldVsn, State, _Extra) ->
 
 %%--------------------------------------------------------------------
 
+%% TODO: This may turn out to be a performance hog when there are
+%% lots of nodes.  We really only need to execute this code on
+%% *one* node, rather than all of them.
+handle_dead_rabbit(Node) ->
+    ok = rabbit_networking:on_node_down(Node),
+    ok = rabbit_amqqueue:on_node_down(Node).
+
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 696dc26588..3908b64692 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -24,7 +24,7 @@
 
 -export([init/4, mainloop/2]).
 
--export([conserve_memory/2, server_properties/0]).
+-export([conserve_memory/2, server_properties/1]).
 
 -export([process_channel_frame/5]). %% used by erlang-client
 
@@ -57,86 +57,6 @@
 
 -define(INFO_KEYS, ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [pid]).
 
-%% connection lifecycle
-%%
-%% all state transitions and terminations are marked with *...*
-%%
-%% The lifecycle begins with: start handshake_timeout timer, *pre-init*
-%%
-%% all states, unless specified otherwise:
-%%   socket error -> *exit*
-%%   socket close -> *throw*
-%%   writer send failure -> *throw*
-%%   forced termination -> *exit*
-%%   handshake_timeout -> *throw*
-%% pre-init:
-%%   receive protocol header -> send connection.start, *starting*
-%% starting:
-%%   receive connection.start_ok -> send connection.tune, *tuning*
-%% tuning:
-%%   receive connection.tune_ok -> start heartbeats, *opening*
-%% opening:
-%%   receive connection.open -> send connection.open_ok, *running*
-%% running:
-%%   receive connection.close ->
-%%     tell channels to terminate gracefully
-%%     if no channels then send connection.close_ok, start
-%%        terminate_connection timer, *closed*
-%%     else *closing*
-%%   forced termination
-%%   -> wait for channels to terminate forcefully, start
-%%      terminate_connection timer, send close, *exit*
-%%   channel exit with hard error
-%%   -> log error, wait for channels to terminate forcefully, start
-%%      terminate_connection timer, send close, *closed*
-%%   channel exit with soft error
-%%   -> log error, mark channel as closing, *running*
-%%   handshake_timeout -> ignore, *running*
-%%   heartbeat timeout -> *throw*
-%%   conserve_memory=true -> *blocking*
-%% blocking:
-%%   conserve_memory=true -> *blocking*
-%%   conserve_memory=false -> *running*
-%%   receive a method frame for a content-bearing method
-%%   -> process, stop receiving, *blocked*
-%%   ...rest same as 'running'
-%% blocked:
-%%   conserve_memory=true -> *blocked*
-%%   conserve_memory=false -> resume receiving, *running*
-%%   ...rest same as 'running'
-%% closing:
-%%   socket close -> *terminate*
-%%   receive connection.close -> send connection.close_ok,
-%%     *closing*
-%%   receive frame -> ignore, *closing*
-%%   handshake_timeout -> ignore, *closing*
-%%   heartbeat timeout -> *throw*
-%%   channel exit with hard error
-%%   -> log error, wait for channels to terminate forcefully, start
-%%      terminate_connection timer, send close, *closed*
-%%   channel exit with soft error
-%%   -> log error, mark channel as closing
-%%      if last channel to exit then send connection.close_ok,
-%%         start terminate_connection timer, *closed*
-%%      else *closing*
-%%   channel exits normally
-%%   -> if last channel to exit then send connection.close_ok,
-%%      start terminate_connection timer, *closed*
-%% closed:
-%%   socket close -> *terminate*
-%%   receive connection.close -> send connection.close_ok,
-%%     *closed*
-%%   receive connection.close_ok -> self() ! terminate_connection,
-%%     *closed*
-%%   receive frame -> ignore, *closed*
-%%   terminate_connection timeout -> *terminate*
-%%   handshake_timeout -> ignore, *closed*
-%%   heartbeat timeout -> *throw*
-%%   channel exit -> log error, *closed*
-%%
-%%
-%% TODO: refactor the code so that the above is obvious
-
 -define(IS_RUNNING(State),
         (State#v1.connection_state =:= running orelse
          State#v1.connection_state =:= blocking orelse
@@ -154,7 +74,8 @@
 -spec(emit_stats/1 :: (pid()) -> 'ok').
 -spec(shutdown/2 :: (pid(), string()) -> 'ok').
 -spec(conserve_memory/2 :: (pid(), boolean()) -> 'ok').
--spec(server_properties/0 :: () -> rabbit_framing:amqp_table()).
+-spec(server_properties/1 :: (rabbit_types:protocol()) ->
+                                  rabbit_framing:amqp_table()).
 
 %% These specs only exists to add no_return() to keep dialyzer happy
 -spec(init/4 :: (pid(), pid(), pid(), rabbit_heartbeat:start_heartbeat_fun())
@@ -213,7 +134,7 @@ conserve_memory(Pid, Conserve) ->
     Pid ! {conserve_memory, Conserve},
     ok.
 
-server_properties() ->
+server_properties(Protocol) ->
     {ok, Product} = application:get_key(rabbit, id),
     {ok, Version} = application:get_key(rabbit, vsn),
 
@@ -224,22 +145,30 @@ server_properties() ->
     %% Normalize the simplifed (2-tuple) and unsimplified (3-tuple) forms
     %% from the config and merge them with the generated built-in properties
     NormalizedConfigServerProps =
-        [case X of
-             {KeyAtom, Value} -> {list_to_binary(atom_to_list(KeyAtom)),
-                                  longstr,
-                                  list_to_binary(Value)};
-             {BinKey, Type, Value} -> {BinKey, Type, Value}
-         end || X <- RawConfigServerProps ++
-                    [{product,     Product},
-                     {version,     Version},
-                     {platform,    "Erlang/OTP"},
-                     {copyright,   ?COPYRIGHT_MESSAGE},
-                     {information, ?INFORMATION_MESSAGE}]],
+        [{<<"capabilities">>, table, server_capabilities(Protocol)} |
+         [case X of
+              {KeyAtom, Value} -> {list_to_binary(atom_to_list(KeyAtom)),
+                                   longstr,
+                                   list_to_binary(Value)};
+              {BinKey, Type, Value} -> {BinKey, Type, Value}
+          end || X <- RawConfigServerProps ++
+                     [{product,     Product},
+                      {version,     Version},
+                      {platform,    "Erlang/OTP"},
+                      {copyright,   ?COPYRIGHT_MESSAGE},
+                      {information, ?INFORMATION_MESSAGE}]]],
 
     %% Filter duplicated properties in favor of config file provided values
     lists:usort(fun ({K1,_,_}, {K2,_,_}) -> K1 =< K2 end,
                 NormalizedConfigServerProps).
 
+server_capabilities(rabbit_framing_amqp_0_9_1) ->
+    [{<<"publisher_confirms">>,         bool, true},
+     {<<"exchange_exchange_bindings">>, bool, true},
+     {<<"basic.nack">>,                 bool, true}];
+server_capabilities(_) ->
+    [].
+
 inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F).
 
 socket_op(Sock, Fun) ->
@@ -256,7 +185,7 @@ start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
                  Sock, SockTransform) ->
     process_flag(trap_exit, true),
     {PeerAddress, PeerPort} = socket_op(Sock, fun rabbit_net:peername/1),
-    PeerAddressS = inet_parse:ntoa(PeerAddress),
+    PeerAddressS = rabbit_misc:ntoab(PeerAddress),
     rabbit_log:info("starting TCP connection ~p from ~s:~p~n",
                     [self(), PeerAddressS, PeerPort]),
     ClientSock = socket_op(Sock, SockTransform),
@@ -322,6 +251,10 @@ mainloop(Deb, State = #v1{parent = Parent, sock= Sock, recv_ref = Ref}) ->
             throw({inet_error, Reason});
         {conserve_memory, Conserve} ->
             mainloop(Deb, internal_conserve_memory(Conserve, State));
+        {channel_closing, ChPid} ->
+            ok = rabbit_channel:ready_for_close(ChPid),
+            channel_cleanup(ChPid),
+            mainloop(Deb, State);
         {'EXIT', Parent, Reason} ->
             terminate(io_lib:format("broker forced connection closure "
                                     "with reason '~w'", [Reason]), State),
@@ -351,7 +284,10 @@ mainloop(Deb, State = #v1{parent = Parent, sock= Sock, recv_ref = Ref}) ->
                     throw({handshake_timeout, State#v1.callback})
             end;
         timeout ->
-            throw({timeout, State#v1.connection_state});
+            case State#v1.connection_state of
+                closed -> mainloop(Deb, State);
+                S      -> throw({timeout, S})
+            end;
         {'$gen_call', From, {shutdown, Explanation}} ->
             {ForceTermination, NewState} = terminate(Explanation, State),
             gen_server:reply(From, ok),
@@ -426,32 +362,32 @@ close_connection(State = #v1{queue_collector = Collector,
     erlang:send_after(TimeoutMillisec, self(), terminate_connection),
     State#v1{connection_state = closed}.
 
-close_channel(Channel, State) ->
-    put({channel, Channel}, closing),
-    State.
-
 handle_dependent_exit(ChPid, Reason, State) ->
     case termination_kind(Reason) of
         controlled ->
-            erase({ch_pid, ChPid}),
+            channel_cleanup(ChPid),
             maybe_close(State);
         uncontrolled ->
             case channel_cleanup(ChPid) of
                 undefined -> exit({abnormal_dependent_exit, ChPid, Reason});
-                Channel   -> maybe_close(
+                Channel   -> rabbit_log:error(
+                               "connection ~p, channel ~p - error:~n~p~n",
+                               [self(), Channel, Reason]),
+                             maybe_close(
                                handle_exception(State, Channel, Reason))
             end
     end.
 
 channel_cleanup(ChPid) ->
     case get({ch_pid, ChPid}) of
-        undefined -> undefined;
-        Channel   -> erase({channel, Channel}),
-                     erase({ch_pid, ChPid}),
-                     Channel
+        undefined       -> undefined;
+        {Channel, MRef} -> erase({channel, Channel}),
+                           erase({ch_pid, ChPid}),
+                           erlang:demonitor(MRef, [flush]),
+                           Channel
     end.
 
-all_channels() -> [ChPid || {{ch_pid, ChPid}, _Channel} <- get()].
+all_channels() -> [ChPid || {{ch_pid, ChPid}, _ChannelMRef} <- get()].
 
 terminate_channels() ->
     NChannels =
@@ -506,8 +442,8 @@ maybe_close(State = #v1{connection_state = closing,
 maybe_close(State) ->
     State.
 
-termination_kind(normal)            -> controlled;
-termination_kind(_)                 -> uncontrolled.
+termination_kind(normal) -> controlled;
+termination_kind(_)      -> uncontrolled.
 
 handle_frame(Type, 0, Payload,
              State = #v1{connection_state = CS,
@@ -543,8 +479,8 @@ handle_frame(Type, Channel, Payload,
                                   Channel, ChPid, FramingState),
                     put({channel, Channel}, {ChPid, NewAState}),
                     case AnalyzedFrame of
-                        {method, 'channel.close', _} ->
-                            erase({channel, Channel}),
+                        {method, 'channel.close_ok', _} ->
+                            channel_cleanup(ChPid),
                             State;
                         {method, MethodName, _} ->
                             case (State#v1.connection_state =:= blocking
@@ -556,25 +492,6 @@ handle_frame(Type, Channel, Payload,
                         _ ->
                             State
                     end;
-                closing ->
-                    %% According to the spec, after sending a
-                    %% channel.close we must ignore all frames except
-                    %% channel.close and channel.close_ok.  In the
-                    %% event of a channel.close, we should send back a
-                    %% channel.close_ok.
-                    case AnalyzedFrame of
-                        {method, 'channel.close_ok', _} ->
-                            erase({channel, Channel});
-                        {method, 'channel.close', _} ->
-                            %% We're already closing this channel, so
-                            %% there's no cleanup to do (notify
-                            %% queues, etc.)
-                            ok = rabbit_writer:internal_send_command(
-                                   State#v1.sock, Channel,
-                                   #'channel.close_ok'{}, Protocol);
-                        _ -> ok
-                    end,
-                    State;
                 undefined ->
                     case ?IS_RUNNING(State) of
                         true  -> send_to_new_channel(
@@ -646,7 +563,7 @@ start_connection({ProtocolMajor, ProtocolMinor, _ProtocolRevision},
     Start = #'connection.start'{
       version_major = ProtocolMajor,
       version_minor = ProtocolMinor,
-      server_properties = server_properties(),
+      server_properties = server_properties(Protocol),
       mechanisms = auth_mechanisms_binary(),
       locales = <<"en_US">> },
     ok = send_on_channel0(Sock, Start, Protocol),
@@ -700,12 +617,18 @@ handle_method0(#'connection.start_ok'{mechanism = Mechanism,
                             connection       = Connection,
                             sock             = Sock}) ->
     AuthMechanism = auth_mechanism_to_module(Mechanism),
+    Capabilities =
+        case rabbit_misc:table_lookup(ClientProperties, <<"capabilities">>) of
+            {table, Capabilities1} -> Capabilities1;
+            _                      -> []
+        end,
     State = State0#v1{auth_mechanism   = AuthMechanism,
                       auth_state       = AuthMechanism:init(Sock),
                       connection_state = securing,
                       connection       =
                           Connection#connection{
-                            client_properties = ClientProperties}},
+                            client_properties = ClientProperties,
+                            capabilities      = Capabilities}},
     auth_phase(Response, State);
 
 handle_method0(#'connection.secure_ok'{response = Response},
@@ -916,10 +839,14 @@ socket_info(Get, Select) ->
     end.
 
 ssl_info(F, Sock) ->
+    %% The first ok form is R14
+    %% The second is R13 - the extra term is exportability (by inspection,
+    %% the docs are wrong)
     case rabbit_net:ssl_info(Sock) of
-        nossl       -> '';
-        {error, _}  -> '';
-        {ok, Info}  -> F(Info)
+        nossl                   -> '';
+        {error, _}              -> '';
+        {ok, {P, {K, C, H}}}    -> F({P, {K, C, H}});
+        {ok, {P, {K, C, H, _}}} -> F({P, {K, C, H}})
     end.
 
 cert_info(F, Sock) ->
@@ -934,19 +861,20 @@ cert_info(F, Sock) ->
 send_to_new_channel(Channel, AnalyzedFrame, State) ->
     #v1{sock = Sock, queue_collector = Collector,
         channel_sup_sup_pid = ChanSupSup,
-        connection = #connection{protocol  = Protocol,
-                                 frame_max = FrameMax,
-                                 user      = User,
-                                 vhost     = VHost}} = State,
+        connection = #connection{protocol     = Protocol,
+                                 frame_max    = FrameMax,
+                                 user         = User,
+                                 vhost        = VHost,
+                                 capabilities = Capabilities}} = State,
     {ok, _ChSupPid, {ChPid, AState}} =
         rabbit_channel_sup_sup:start_channel(
-          ChanSupSup, {Protocol, Sock, Channel, FrameMax,
-                       self(), User, VHost, Collector}),
-    erlang:monitor(process, ChPid),
+          ChanSupSup, {tcp, Sock, Channel, FrameMax, self(), Protocol, User,
+                       VHost, Capabilities, Collector}),
+    MRef = erlang:monitor(process, ChPid),
     NewAState = process_channel_frame(AnalyzedFrame, self(),
                                       Channel, ChPid, AState),
     put({channel, Channel}, {ChPid, NewAState}),
-    put({ch_pid, ChPid}, Channel),
+    put({ch_pid, ChPid}, {Channel, MRef}),
     State.
 
 process_channel_frame(Frame, ErrPid, Channel, ChPid, AState) ->
@@ -962,29 +890,20 @@ process_channel_frame(Frame, ErrPid, Channel, ChPid, AState) ->
                                             AState
     end.
 
-log_channel_error(ConnectionState, Channel, Reason) ->
-    rabbit_log:error("connection ~p (~p), channel ~p - error:~n~p~n",
-                     [self(), ConnectionState, Channel, Reason]).
-
-handle_exception(State = #v1{connection_state = closed}, Channel, Reason) ->
-    log_channel_error(closed, Channel, Reason),
+handle_exception(State = #v1{connection_state = closed}, _Channel, _Reason) ->
     State;
-handle_exception(State = #v1{connection_state = CS}, Channel, Reason) ->
-    log_channel_error(CS, Channel, Reason),
+handle_exception(State, Channel, Reason) ->
     send_exception(State, Channel, Reason).
 
 send_exception(State = #v1{connection = #connection{protocol = Protocol}},
                Channel, Reason) ->
-    {ShouldClose, CloseChannel, CloseMethod} =
+    {0, CloseMethod} =
         rabbit_binary_generator:map_exception(Channel, Reason, Protocol),
-    NewState = case ShouldClose of
-                   true  -> terminate_channels(),
-                            close_connection(State);
-                   false -> close_channel(Channel, State)
-               end,
+    terminate_channels(),
+    State1 = close_connection(State),
     ok = rabbit_writer:internal_send_command(
-           NewState#v1.sock, CloseChannel, CloseMethod, Protocol),
-    NewState.
+           State1#v1.sock, 0, CloseMethod, Protocol),
+    State1.
 
 internal_emit_stats(State = #v1{stats_timer = StatsTimer}) ->
     rabbit_event:notify(connection_stats, infos(?STATISTICS_KEYS, State)),
diff --git a/src/rabbit_registry.erl b/src/rabbit_registry.erl
index 795413aa5c..9821ae7b86 100644
--- a/src/rabbit_registry.erl
+++ b/src/rabbit_registry.erl
@@ -48,7 +48,7 @@ start_link() ->
 %%---------------------------------------------------------------------------
 
 register(Class, TypeName, ModuleName) ->
-    gen_server:call(?SERVER, {register, Class, TypeName, ModuleName}).
+    gen_server:call(?SERVER, {register, Class, TypeName, ModuleName}, infinity).
 
 %% This is used with user-supplied arguments (e.g., on exchange
 %% declare), so we restrict it to existing atoms only.  This means it
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 644c4f9662..3db393ef0b 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -26,6 +26,7 @@
 
 -define(PERSISTENT_MSG_STORE, msg_store_persistent).
 -define(TRANSIENT_MSG_STORE,  msg_store_transient).
+-define(CLEANUP_QUEUE_NAME, <<"cleanup-queue">>).
 
 test_content_prop_roundtrip(Datum, Binary) ->
     Types =  [element(1, E) || E <- Datum],
@@ -81,20 +82,24 @@ run_cluster_dependent_tests(SecondaryNode) ->
     io:format("Running cluster dependent tests with node ~p~n", [SecondaryNode]),
     passed = test_delegates_async(SecondaryNode),
     passed = test_delegates_sync(SecondaryNode),
+    passed = test_queue_cleanup(SecondaryNode),
+    passed = test_declare_on_dead_queue(SecondaryNode),
 
     %% we now run the tests remotely, so that code coverage on the
     %% local node picks up more of the delegate
     Node = node(),
     Self = self(),
     Remote = spawn(SecondaryNode,
-                   fun () -> A = test_delegates_async(Node),
-                             B = test_delegates_sync(Node),
-                             Self ! {self(), {A, B}}
+                   fun () -> Rs = [ test_delegates_async(Node),
+                                    test_delegates_sync(Node),
+                                    test_queue_cleanup(Node),
+                                    test_declare_on_dead_queue(Node) ],
+                             Self ! {self(), Rs}
                    end),
     receive
         {Remote, Result} ->
-            Result = {passed, passed}
-    after 2000 ->
+            Result = lists:duplicate(length(Result), passed)
+    after 30000 ->
             throw(timeout)
     end,
 
@@ -1015,9 +1020,9 @@ test_user_management() ->
 test_server_status() ->
     %% create a few things so there is some useful information to list
     Writer = spawn(fun () -> receive shutdown -> ok end end),
-    {ok, Ch} = rabbit_channel:start_link(1, self(), Writer,
-                                         user(<<"user">>), <<"/">>, self(),
-                                         fun (_) -> {ok, self()} end),
+    {ok, Ch} = rabbit_channel:start_link(
+                 1, self(), Writer, rabbit_framing_amqp_0_9_1, user(<<"user">>),
+                 <<"/">>, [], self(), fun (_) -> {ok, self()} end),
     [Q, Q2] = [Queue || Name <- [<<"foo">>, <<"bar">>],
                         {new, Queue = #amqqueue{}} <-
                             [rabbit_amqqueue:declare(
@@ -1075,9 +1080,9 @@ test_server_status() ->
 test_spawn(Receiver) ->
     Me = self(),
     Writer = spawn(fun () -> Receiver(Me) end),
-    {ok, Ch} = rabbit_channel:start_link(1, Me, Writer,
-                                         user(<<"guest">>), <<"/">>, self(),
-                                         fun (_) -> {ok, self()} end),
+    {ok, Ch} = rabbit_channel:start_link(
+                 1, Me, Writer, rabbit_framing_amqp_0_9_1, user(<<"guest">>),
+                 <<"/">>, [], self(), fun (_) -> {ok, self()} end),
     ok = rabbit_channel:do(Ch, #'channel.open'{}),
     receive #'channel.open_ok'{} -> ok
     after 1000 -> throw(failed_to_receive_channel_open_ok)
@@ -1229,7 +1234,7 @@ must_exit(Fun) ->
     end.
 
 test_delegates_sync(SecondaryNode) ->
-    Sender = fun (Pid) -> gen_server:call(Pid, invoked) end,
+    Sender = fun (Pid) -> gen_server:call(Pid, invoked, infinity) end,
     BadSender = fun (_Pid) -> exit(exception) end,
 
     Responder = make_responder(fun ({'$gen_call', From, invoked}) ->
@@ -1279,6 +1284,61 @@ test_delegates_sync(SecondaryNode) ->
 
     passed.
 
+test_queue_cleanup_receiver(Pid) ->
+    receive
+        shutdown ->
+            ok;
+        {send_command, Method} ->
+            Pid ! Method,
+            test_queue_cleanup_receiver(Pid)
+    end.
+
+
+test_queue_cleanup(_SecondaryNode) ->
+    {_Writer, Ch} = test_spawn(fun test_queue_cleanup_receiver/1),
+    rabbit_channel:do(Ch, #'queue.declare'{ queue = ?CLEANUP_QUEUE_NAME }),
+    receive #'queue.declare_ok'{queue = ?CLEANUP_QUEUE_NAME} ->
+            ok
+    after 1000 -> throw(failed_to_receive_queue_declare_ok)
+    end,
+    rabbit:stop(),
+    rabbit:start(),
+    rabbit_channel:do(Ch, #'queue.declare'{ passive = true,
+                                            queue   = ?CLEANUP_QUEUE_NAME }),
+    receive
+        #'channel.close'{reply_code = 404} ->
+            ok
+    after 2000 ->
+            throw(failed_to_receive_channel_exit)
+    end,
+    passed.
+
+test_declare_on_dead_queue(SecondaryNode) ->
+    QueueName = rabbit_misc:r(<<"/">>, queue, ?CLEANUP_QUEUE_NAME),
+    Self = self(),
+    Pid = spawn(SecondaryNode,
+                fun () ->
+                        {new, #amqqueue{name = QueueName, pid = QPid}} =
+                            rabbit_amqqueue:declare(QueueName, false, false, [],
+                                                    none),
+                        exit(QPid, kill),
+                        Self ! {self(), killed, QPid}
+                end),
+    receive
+        {Pid, killed, QPid} ->
+            {existing, #amqqueue{name = QueueName,
+                                 pid = QPid}} =
+                rabbit_amqqueue:declare(QueueName, false, false, [], none),
+            false = rabbit_misc:is_process_alive(QPid),
+            {new, Q} = rabbit_amqqueue:declare(QueueName, false, false, [],
+                                               none),
+            true = rabbit_misc:is_process_alive(Q#amqqueue.pid),
+            {ok, 0} = rabbit_amqqueue:delete(Q, false, false),
+            passed
+    after 2000 ->
+            throw(failed_to_create_and_kill_queue)
+    end.
+
 %---------------------------------------------------------------------
 
 control_action(Command, Args) ->
@@ -2142,9 +2202,11 @@ test_configurable_server_properties() ->
     BuiltInPropNames = [<<"product">>, <<"version">>, <<"platform">>,
                         <<"copyright">>, <<"information">>],
 
+    Protocol = rabbit_framing_amqp_0_9_1,
+
     %% Verify that the built-in properties are initially present
-    ActualPropNames = [Key ||
-                         {Key, longstr, _} <- rabbit_reader:server_properties()],
+    ActualPropNames = [Key || {Key, longstr, _} <-
+                                  rabbit_reader:server_properties(Protocol)],
     true = lists:all(fun (X) -> lists:member(X, ActualPropNames) end,
                      BuiltInPropNames),
 
@@ -2155,9 +2217,10 @@ test_configurable_server_properties() ->
     ConsProp = fun (X) -> application:set_env(rabbit,
                                               server_properties,
                                               [X | ServerProperties]) end,
-    IsPropPresent = fun (X) -> lists:member(X,
-                                            rabbit_reader:server_properties())
-                    end,
+    IsPropPresent =
+        fun (X) ->
+                lists:member(X, rabbit_reader:server_properties(Protocol))
+        end,
 
     %% Add a wholly new property of the simplified {KeyAtom, StringValue} form
     NewSimplifiedProperty = {NewHareKey, NewHareVal} = {hare, "soup"},
@@ -2180,7 +2243,7 @@ test_configurable_server_properties() ->
     {BinNewVerKey, BinNewVerVal} = {list_to_binary(atom_to_list(NewVerKey)),
                                     list_to_binary(NewVerVal)},
     ConsProp(NewVersion),
-    ClobberedServerProps = rabbit_reader:server_properties(),
+    ClobberedServerProps = rabbit_reader:server_properties(Protocol),
     %% Is the clobbering insert present?
     true = IsPropPresent({BinNewVerKey, longstr, BinNewVerVal}),
     %% Is the clobbering insert the only thing with the clobbering key?
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index f39bc96426..7142d56072 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -281,12 +281,11 @@
 -record(sync, { acks_persistent, acks_all, pubs, funs }).
 
 %% When we discover, on publish, that we should write some indices to
-%% disk for some betas, the RAM_INDEX_BATCH_SIZE sets the number of
-%% betas that we must be due to write indices for before we do any
-%% work at all. This is both a minimum and a maximum - we don't write
-%% fewer than RAM_INDEX_BATCH_SIZE indices out in one go, and we don't
-%% write more - we can always come back on the next publish to do
-%% more.
+%% disk for some betas, the IO_BATCH_SIZE sets the number of betas
+%% that we must be due to write indices for before we do any work at
+%% all. This is both a minimum and a maximum - we don't write fewer
+%% than IO_BATCH_SIZE indices out in one go, and we don't write more -
+%% we can always come back on the next publish to do more.
 -define(IO_BATCH_SIZE, 64).
 -define(PERSISTENT_MSG_STORE, msg_store_persistent).
 -define(TRANSIENT_MSG_STORE,  msg_store_transient).
@@ -299,7 +298,7 @@
 
 -type(timestamp() :: {non_neg_integer(), non_neg_integer(), non_neg_integer()}).
 -type(seq_id()  :: non_neg_integer()).
--type(ack()     :: seq_id() | 'blank_ack').
+-type(ack()     :: seq_id()).
 
 -type(rates() :: #rates { egress      :: {timestamp(), non_neg_integer()},
                           ingress     :: {timestamp(), non_neg_integer()},
@@ -509,7 +508,7 @@ publish(Msg, MsgProps, State) ->
 publish_delivered(false, #basic_message { guid = Guid },
                   _MsgProps, State = #vqstate { len = 0 }) ->
     blind_confirm(self(), gb_sets:singleton(Guid)),
-    {blank_ack, a(State)};
+    {undefined, a(State)};
 publish_delivered(true, Msg = #basic_message { is_persistent = IsPersistent,
                                                guid = Guid },
                   MsgProps = #message_properties {
@@ -628,7 +627,7 @@ internal_fetch(AckRequired, MsgStatus = #msg_status {
                                             MsgStatus #msg_status {
                                               is_delivered = true }, State),
                                  {SeqId, StateN};
-                        false -> {blank_ack, State}
+                        false -> {undefined, State}
                     end,
 
     PCount1 = PCount - one_if(IsPersistent andalso not AckRequired),
@@ -897,7 +896,7 @@ cons_if(true,   E, L) -> [E | L];
 cons_if(false, _E, L) -> L.
 
 gb_sets_maybe_insert(false, _Val, Set) -> Set;
-%% when requeueing, we re-add a guid to the unconfimred set
+%% when requeueing, we re-add a guid to the unconfirmed set
 gb_sets_maybe_insert(true,  Val,  Set) -> gb_sets:add(Val, Set).
 
 msg_status(IsPersistent, SeqId, Msg = #basic_message { guid = Guid },
diff --git a/src/supervisor2.erl b/src/supervisor2.erl
index 18e2bdadb9..1a240856ce 100644
--- a/src/supervisor2.erl
+++ b/src/supervisor2.erl
@@ -359,8 +359,8 @@ handle_cast({delayed_restart, {RestartType, Reason, Child}}, State)
     {noreply, NState};
 handle_cast({delayed_restart, {RestartType, Reason, Child}}, State) ->
     case get_child(Child#child.name, State) of
-        {value, Child} ->
-            {ok, NState} = do_restart(RestartType, Reason, Child, State),
+        {value, Child1} ->
+            {ok, NState} = do_restart(RestartType, Reason, Child1, State),
             {noreply, NState};
         _ ->
             {noreply, State}
@@ -539,7 +539,7 @@ do_restart({RestartType, Delay}, Reason, Child, State) ->
             {ok, _TRef} = timer:apply_after(
                             trunc(Delay*1000), ?MODULE, delayed_restart,
                             [self(), {{RestartType, Delay}, Reason, Child}]),
-            {ok, NState}
+            {ok, state_del_child(Child, NState)}
     end;
 do_restart(permanent, Reason, Child, State) ->
     report_error(child_terminated, Reason, Child, State#state.name),
diff --git a/src/tcp_acceptor.erl b/src/tcp_acceptor.erl
index 194389e32a..0d50683db7 100644
--- a/src/tcp_acceptor.erl
+++ b/src/tcp_acceptor.erl
@@ -59,8 +59,8 @@ handle_info({inet_async, LSock, Ref, {ok, Sock}},
         {Address, Port}         = inet_op(fun () -> inet:sockname(LSock) end),
         {PeerAddress, PeerPort} = inet_op(fun () -> inet:peername(Sock) end),
         error_logger:info_msg("accepted TCP connection on ~s:~p from ~s:~p~n",
-                              [inet_parse:ntoa(Address), Port,
-                               inet_parse:ntoa(PeerAddress), PeerPort]),
+                              [rabbit_misc:ntoab(Address), Port,
+                               rabbit_misc:ntoab(PeerAddress), PeerPort]),
         %% In the event that somebody floods us with connections we can spew
         %% the above message at error_logger faster than it can keep up.
         %% So error_logger's mailbox grows unbounded until we eat all the
diff --git a/src/tcp_listener.erl b/src/tcp_listener.erl
index b1bfcafcdb..cd64696904 100644
--- a/src/tcp_listener.erl
+++ b/src/tcp_listener.erl
@@ -50,8 +50,9 @@ init({IPAddress, Port, SocketOpts,
                           end,
                           lists:duplicate(ConcurrentAcceptorCount, dummy)),
             {ok, {LIPAddress, LPort}} = inet:sockname(LSock),
-            error_logger:info_msg("started ~s on ~s:~p~n",
-                                  [Label, inet_parse:ntoa(LIPAddress), LPort]),
+            error_logger:info_msg(
+              "started ~s on ~s:~p~n",
+              [Label, rabbit_misc:ntoab(LIPAddress), LPort]),
             apply(M, F, A ++ [IPAddress, Port]),
             {ok, #state{sock = LSock,
                         on_startup = OnStartup, on_shutdown = OnShutdown,
@@ -59,7 +60,7 @@ init({IPAddress, Port, SocketOpts,
         {error, Reason} ->
             error_logger:error_msg(
               "failed to start ~s on ~s:~p - ~p~n",
-              [Label, inet_parse:ntoa(IPAddress), Port, Reason]),
+              [Label, rabbit_misc:ntoab(IPAddress), Port, Reason]),
             {stop, {cannot_listen, IPAddress, Port, Reason}}
     end.
 
@@ -76,7 +77,7 @@ terminate(_Reason, #state{sock=LSock, on_shutdown = {M,F,A}, label=Label}) ->
     {ok, {IPAddress, Port}} = inet:sockname(LSock),
     gen_tcp:close(LSock),
     error_logger:info_msg("stopped ~s on ~s:~p~n",
-                          [Label, inet_parse:ntoa(IPAddress), Port]),
+                          [Label, rabbit_misc:ntoab(IPAddress), Port]),
     apply(M, F, A ++ [IPAddress, Port]).
 
 code_change(_OldVsn, State, _Extra) ->
diff --git a/src/test_sup.erl b/src/test_sup.erl
index 76be63d0cf..b4df1fd042 100644
--- a/src/test_sup.erl
+++ b/src/test_sup.erl
@@ -59,19 +59,21 @@ start_child() ->
 
 ping_child(SupPid) ->
     Ref = make_ref(),
-    get_child_pid(SupPid) ! {ping, Ref, self()},
+    with_child_pid(SupPid, fun(ChildPid) -> ChildPid ! {ping, Ref, self()} end),
     receive {pong, Ref} -> ok
     after 1000          -> timeout
     end.
 
 exit_child(SupPid) ->
-    true = exit(get_child_pid(SupPid), abnormal),
+    with_child_pid(SupPid, fun(ChildPid) -> exit(ChildPid, abnormal) end),
     ok.
 
-get_child_pid(SupPid) ->
-    [{_Id, ChildPid, worker, [test_sup]}] =
-        supervisor2:which_children(SupPid),
-    ChildPid.
+with_child_pid(SupPid, Fun) ->
+    case supervisor2:which_children(SupPid) of
+        [{_Id, undefined, worker, [test_sup]}] -> ok;
+        [{_Id,  ChildPid, worker, [test_sup]}] -> Fun(ChildPid);
+        []                                     -> ok
+    end.
 
 run_child() ->
     receive {ping, Ref, Pid} -> Pid ! {pong, Ref},