Mereg default into bug23566.

42 files changed, 884 insertions, 1236 deletions

diff --git a/Makefile b/Makefile
index d3f052f6..e0d5744c 100644
--- a/Makefile
+++ b/Makefile
@@ -178,9 +178,6 @@ start-rabbit-on-node: all
 stop-rabbit-on-node: all
 	echo "rabbit:stop()." | $(ERL_CALL)
 
-force-snapshot: all
-	echo "rabbit_persister:force_snapshot()." | $(ERL_CALL)
-
 set-memory-alarm: all
 	echo "alarm_handler:set_alarm({vm_memory_high_watermark, []})." | \
 	$(ERL_CALL)
diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index 6b02abe4..3c45217d 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -1005,6 +1005,10 @@
                 <listitem><para>Version of the AMQP protocol in use (currently one of <command>{0,9,1}</command> or <command>{0,8,0}</command>). Note that if a client requests an AMQP 0-9 connection, we treat it as AMQP 0-9-1.</para></listitem>
               </varlistentry>
               <varlistentry>
+                <term>auth_mechanism</term>
+                <listitem><para>SASL authentication mechanism used, such as <command>PLAIN</command>.</para></listitem>
+              </varlistentry>
+              <varlistentry>
                 <term>user</term>
                 <listitem><para>Username associated with the connection.</para></listitem>
               </varlistentry>
diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index 17d05a99..3888f198 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -6,7 +6,6 @@
   {registered, [rabbit_amqqueue_sup,
                 rabbit_log,
                 rabbit_node_monitor,
-                rabbit_persister,
                 rabbit_router,
                 rabbit_sup,
                 rabbit_tcp_client_sup]},
@@ -31,4 +30,5 @@
          {default_permissions, [<<".*">>, <<".*">>, <<".*">>]},
          {cluster_nodes, []},
          {server_properties, []},
-         {collect_statistics, none}]}]}.
+         {collect_statistics, none},
+         {auth_mechanisms, ['PLAIN', 'AMQPLAIN']} ]} ]}.
diff --git a/include/rabbit_auth_mechanism_spec.hrl b/include/rabbit_auth_mechanism_spec.hrl
new file mode 100644
index 00000000..93aa40bd
--- /dev/null
+++ b/include/rabbit_auth_mechanism_spec.hrl
@@ -0,0 +1,41 @@
+%%   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 Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+-ifdef(use_specs).
+
+-spec(description/0 :: () -> [{atom(), any()}]).
+-spec(init/1 :: (rabbit_net:socket()) -> any()).
+-spec(handle_response/2 :: (binary(), any()) ->
+                                {'ok', rabbit_types:user()} |
+                                {'challenge', binary(), any()} |
+                                {'protocol_error', string(), [any()]} |
+                                {'refused', rabbit_access_control:username()}).
+
+-endif.
diff --git a/scripts/rabbitmq-env b/scripts/rabbitmq-env
index 36734874..8cb470d0 100755
--- a/scripts/rabbitmq-env
+++ b/scripts/rabbitmq-env
@@ -48,6 +48,8 @@ done
 
 SCRIPT_DIR=`dirname $SCRIPT_PATH`
 RABBITMQ_HOME="${SCRIPT_DIR}/.."
+[ "x" = "x$HOSTNAME" ] && HOSTNAME=`env hostname`
+NODENAME=rabbit@${HOSTNAME%%.*}
 
 # Load configuration from the rabbitmq.conf file
 [ -f /etc/rabbitmq/rabbitmq.conf ] && . /etc/rabbitmq/rabbitmq.conf
diff --git a/scripts/rabbitmq-multi b/scripts/rabbitmq-multi
index 59050692..33883702 100755
--- a/scripts/rabbitmq-multi
+++ b/scripts/rabbitmq-multi
@@ -29,8 +29,7 @@
 ##
 ##   Contributor(s): ______________________________________.
 ##
-[ "x" = "x$HOSTNAME" ] && HOSTNAME=`env hostname -s`
-NODENAME=rabbit@${HOSTNAME%%.*}
+
 SCRIPT_HOME=$(dirname $0)
 PIDS_FILE=/var/lib/rabbitmq/pids
 MULTI_ERL_ARGS=
diff --git a/scripts/rabbitmq-server b/scripts/rabbitmq-server
index c5d883c3..4155b31d 100755
--- a/scripts/rabbitmq-server
+++ b/scripts/rabbitmq-server
@@ -30,8 +30,6 @@
 ##   Contributor(s): ______________________________________.
 ##
 
-[ "x" = "x$HOSTNAME" ] && HOSTNAME=`env hostname -s`
-NODENAME=rabbit@${HOSTNAME%%.*}
 SERVER_ERL_ARGS="+K true +A30 +P 1048576 \
 -kernel inet_default_listen_options [{nodelay,true}] \
 -kernel inet_default_connect_options [{nodelay,true}]"
@@ -91,8 +89,9 @@ if [ "x" = "x$RABBITMQ_NODE_ONLY" ]; then
 	-pa "$RABBITMQ_EBIN_ROOT" \
 	-noinput \
 	-hidden \
-	-s rabbit_plugin_activator \
-	-extra "$RABBITMQ_PLUGINS_DIR" "${RABBITMQ_PLUGINS_EXPAND_DIR}"
+	-s rabbit_prelaunch \
+	-sname rabbitmqprelaunch$$ \
+	-extra "$RABBITMQ_PLUGINS_DIR" "${RABBITMQ_PLUGINS_EXPAND_DIR}" "${RABBITMQ_NODENAME}"
     then
 	RABBITMQ_BOOT_FILE="${RABBITMQ_PLUGINS_EXPAND_DIR}/rabbit"
 	RABBITMQ_EBIN_PATH=""
diff --git a/scripts/rabbitmq-server.bat b/scripts/rabbitmq-server.bat
index 94180de9..52a250c6 100644
--- a/scripts/rabbitmq-server.bat
+++ b/scripts/rabbitmq-server.bat
@@ -117,13 +117,14 @@ set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin
 "!ERLANG_HOME!\bin\erl.exe" ^

 -pa "!RABBITMQ_EBIN_ROOT!" ^

 -noinput -hidden ^

--s rabbit_plugin_activator ^

+-s rabbit_prelaunch ^

+-sname rabbitmqprelaunch%RANDOM% ^

 -extra "!RABBITMQ_PLUGINS_DIR:\=/!" ^

-       "!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!"

+       "!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!" ^

+       "!RABBITMQ_NODENAME!"

 

 set RABBITMQ_BOOT_FILE=!RABBITMQ_PLUGINS_EXPAND_DIR!\rabbit

-if not exist "!RABBITMQ_BOOT_FILE!.boot" (

-    echo Custom Boot File "!RABBITMQ_BOOT_FILE!.boot" is missing.

+if ERRORLEVEL 1 (

     exit /B 1

 )

 

diff --git a/scripts/rabbitmq-service.bat b/scripts/rabbitmq-service.bat
index 2c96b6fd..d2592931 100644
--- a/scripts/rabbitmq-service.bat
+++ b/scripts/rabbitmq-service.bat
@@ -186,13 +186,13 @@ set RABBITMQ_EBIN_ROOT=!TDP0!..\ebin
 "!ERLANG_HOME!\bin\erl.exe" ^

 -pa "!RABBITMQ_EBIN_ROOT!" ^

 -noinput -hidden ^

--s rabbit_plugin_activator ^

+-s rabbit_prelaunch ^

 -extra "!RABBITMQ_PLUGINS_DIR:\=/!" ^

-       "!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!"

+       "!RABBITMQ_PLUGINS_EXPAND_DIR:\=/!" ^

+       ""

 

 set RABBITMQ_BOOT_FILE=!RABBITMQ_PLUGINS_EXPAND_DIR!\rabbit

-if not exist "!RABBITMQ_BOOT_FILE!.boot" (

-    echo Custom Boot File "!RABBITMQ_BOOT_FILE!.boot" is missing.

+if ERRORLEVEL 1 (

     exit /B 1

 )

 

diff --git a/scripts/rabbitmqctl b/scripts/rabbitmqctl
index 76ce25fd..56cff891 100755
--- a/scripts/rabbitmqctl
+++ b/scripts/rabbitmqctl
@@ -30,9 +30,6 @@
 ##   Contributor(s): ______________________________________.
 ##
 
-[ "x" = "x$HOSTNAME" ] && HOSTNAME=`env hostname -s`
-NODENAME=rabbit@${HOSTNAME%%.*}
-
 . `dirname $0`/rabbitmq-env
 
 [ "x" = "x$RABBITMQ_NODENAME" ] && RABBITMQ_NODENAME=${NODENAME}
diff --git a/src/rabbit.erl b/src/rabbit.erl
index a619ef36..2ebfdecf 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -69,10 +69,10 @@
 -rabbit_boot_step({external_infrastructure,
                    [{description, "external infrastructure ready"}]}).
 
--rabbit_boot_step({rabbit_exchange_type_registry,
-                   [{description, "exchange type registry"},
+-rabbit_boot_step({rabbit_registry,
+                   [{description, "plugin registry"},
                     {mfa,         {rabbit_sup, start_child,
-                                   [rabbit_exchange_type_registry]}},
+                                   [rabbit_registry]}},
                     {requires,    external_infrastructure},
                     {enables,     kernel_ready}]}).
 
@@ -170,12 +170,6 @@
 
 %%---------------------------------------------------------------------------
 
--import(application).
--import(mnesia).
--import(lists).
--import(inet).
--import(gen_tcp).
-
 -include("rabbit_framing.hrl").
 -include("rabbit.hrl").
 
diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index 8288fbed..c15a4de9 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -33,7 +33,7 @@
 -include_lib("stdlib/include/qlc.hrl").
 -include("rabbit.hrl").
 
--export([check_login/2, user_pass_login/2, check_user_pass_login/2,
+-export([user_pass_login/2, check_user_pass_login/2, make_salt/0,
          check_vhost_access/2, check_resource_access/3]).
 -export([add_user/2, delete_user/1, change_password/2, set_admin/1,
          clear_admin/1, list_users/0, lookup_user/1]).
@@ -54,15 +54,13 @@
 -type(password() :: binary()).
 -type(password_hash() :: binary()).
 -type(regexp() :: binary()).
--spec(check_login/2 ::
-        (binary(), binary()) -> rabbit_types:user() |
-                                rabbit_types:channel_exit()).
 -spec(user_pass_login/2 ::
         (username(), password())
         -> rabbit_types:user() | rabbit_types:channel_exit()).
 -spec(check_user_pass_login/2 ::
         (username(), password())
-        -> {'ok', rabbit_types:user()} | 'refused').
+        -> {'ok', rabbit_types:user()} | {'refused', username()}).
+-spec(make_salt/0 :: () -> binary()).
 -spec(check_vhost_access/2 ::
         (rabbit_types:user(), rabbit_types:vhost())
         -> 'ok' | rabbit_types:channel_exit()).
@@ -100,61 +98,27 @@
 
 %%----------------------------------------------------------------------------
 
-%% SASL PLAIN, as used by the Qpid Java client and our clients. Also,
-%% apparently, by OpenAMQ.
-check_login(<<"PLAIN">>, Response) ->
-    case re:run(Response, "\\0([^\\0]*)", [{capture, all_but_first, binary},
-                                           global]) of
-        {match, [[User],[Pass]]} ->
-            user_pass_login(User, Pass);
-        _ ->
-            rabbit_misc:protocol_error(
-              access_refused, "login refused, response ~p invalid",
-              [Response])
-    end;
-
-%% AMQPLAIN, as used by Qpid Python test suite. The 0-8 spec actually
-%% defines this as PLAIN, but in 0-9 that definition is gone, instead
-%% referring generically to "SASL security mechanism", i.e. the above.
-check_login(<<"AMQPLAIN">>, Response) ->
-    LoginTable = rabbit_binary_parser:parse_table(Response),
-    case {lists:keysearch(<<"LOGIN">>, 1, LoginTable),
-          lists:keysearch(<<"PASSWORD">>, 1, LoginTable)} of
-        {{value, {_, longstr, User}},
-         {value, {_, longstr, Pass}}} ->
-            user_pass_login(User, Pass);
-        _ ->
-            %% Is this an information leak?
-            rabbit_misc:protocol_error(
-              access_refused,
-              "AMQPPLAIN auth info ~w is missing LOGIN or PASSWORD field",
-              [LoginTable])
-    end;
-
-check_login(Mechanism, _Response) ->
-    rabbit_misc:protocol_error(
-      access_refused, "unsupported authentication mechanism '~s'",
-      [Mechanism]).
-
 user_pass_login(User, Pass) ->
     ?LOGDEBUG("Login with user ~p pass ~p~n", [User, Pass]),
     case check_user_pass_login(User, Pass) of
-        refused ->
+        {refused, _} ->
             rabbit_misc:protocol_error(
               access_refused, "login refused for user '~s'", [User]);
         {ok, U} ->
             U
     end.
 
-check_user_pass_login(User, Pass) ->
-    case lookup_user(User) of
-        {ok, U} ->
-            case check_password(Pass, U#user.password_hash) of
-                true -> {ok, U};
-                _    -> refused
+check_user_pass_login(Username, Pass) ->
+    Refused = {refused, io_lib:format("user '~s' - invalid credentials",
+                                      [Username])},
+    case lookup_user(Username) of
+        {ok, User} ->
+            case check_password(Pass, User#user.password_hash) of
+                true -> {ok, User};
+                _    -> Refused
             end;
         {error, not_found} ->
-            refused
+            Refused
     end.
 
 internal_lookup_vhost_access(Username, VHostPath) ->
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 70d8f2dd..775c631d 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -49,11 +49,6 @@
 -export([commit_all/3, rollback_all/3, notify_down_all/2, limit_all/3]).
 -export([on_node_down/1]).
 
--import(mnesia).
--import(gen_server2).
--import(lists).
--import(queue).
-
 -include("rabbit.hrl").
 -include_lib("stdlib/include/qlc.hrl").
 
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 25859c22..78bb6835 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -48,10 +48,6 @@
          handle_info/2, handle_pre_hibernate/1, prioritise_call/3,
          prioritise_cast/2, prioritise_info/2]).
 
--import(queue).
--import(erlang).
--import(lists).
-
 % Queue's state
 -record(q, {q,
             exclusive_consumer,
diff --git a/src/rabbit_auth_mechanism.erl b/src/rabbit_auth_mechanism.erl
new file mode 100644
index 00000000..1258cb8d
--- /dev/null
+++ b/src/rabbit_auth_mechanism.erl
@@ -0,0 +1,57 @@
+%%   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 Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_auth_mechanism).
+
+-export([behaviour_info/1]).
+
+behaviour_info(callbacks) ->
+    [
+     %% A description.
+     {description, 0},
+
+     %% Called before authentication starts. Should create a state
+     %% object to be passed through all the stages of authentication.
+     {init, 1},
+
+     %% Handle a stage of authentication. Possible responses:
+     %% {ok, User}
+     %%     Authentication succeeded, and here's the user record.
+     %% {challenge, Challenge, NextState}
+     %%     Another round is needed. Here's the state I want next time.
+     %% {protocol_error, Msg, Args}
+     %%     Client got the protocol wrong. Log and die.
+     %% {refused, Username}
+     %%     Client failed authentication. Log and die.
+     {handle_response, 2}
+    ];
+behaviour_info(_Other) ->
+    undefined.
diff --git a/src/rabbit_auth_mechanism_amqplain.erl b/src/rabbit_auth_mechanism_amqplain.erl
new file mode 100644
index 00000000..5d51d904
--- /dev/null
+++ b/src/rabbit_auth_mechanism_amqplain.erl
@@ -0,0 +1,70 @@
+%%   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 Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_auth_mechanism_amqplain).
+-include("rabbit.hrl").
+
+-behaviour(rabbit_auth_mechanism).
+
+-export([description/0, init/1, handle_response/2]).
+
+-include("rabbit_auth_mechanism_spec.hrl").
+
+-rabbit_boot_step({?MODULE,
+                   [{description, "auth mechanism amqplain"},
+                    {mfa,         {rabbit_registry, register,
+                                   [auth_mechanism, <<"AMQPLAIN">>, ?MODULE]}},
+                    {requires,    rabbit_registry},
+                    {enables,     kernel_ready}]}).
+
+%% AMQPLAIN, as used by Qpid Python test suite. The 0-8 spec actually
+%% defines this as PLAIN, but in 0-9 that definition is gone, instead
+%% referring generically to "SASL security mechanism", i.e. the above.
+
+description() ->
+    [{name, <<"AMQPLAIN">>},
+     {description, <<"QPid AMQPLAIN mechanism">>}].
+
+init(_Sock) ->
+    [].
+
+handle_response(Response, _State) ->
+    LoginTable = rabbit_binary_parser:parse_table(Response),
+    case {lists:keysearch(<<"LOGIN">>, 1, LoginTable),
+          lists:keysearch(<<"PASSWORD">>, 1, LoginTable)} of
+        {{value, {_, longstr, User}},
+         {value, {_, longstr, Pass}}} ->
+            rabbit_access_control:check_user_pass_login(User, Pass);
+        _ ->
+            {protocol_error,
+             "AMQPLAIN auth info ~w is missing LOGIN or PASSWORD field",
+              [LoginTable]}
+    end.
diff --git a/src/rabbit_auth_mechanism_cr_demo.erl b/src/rabbit_auth_mechanism_cr_demo.erl
new file mode 100644
index 00000000..67665928
--- /dev/null
+++ b/src/rabbit_auth_mechanism_cr_demo.erl
@@ -0,0 +1,74 @@
+%%   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 Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_auth_mechanism_cr_demo).
+-include("rabbit.hrl").
+
+-behaviour(rabbit_auth_mechanism).
+
+-export([description/0, init/1, handle_response/2]).
+
+-include("rabbit_auth_mechanism_spec.hrl").
+
+-rabbit_boot_step({?MODULE,
+                   [{description, "auth mechanism cr-demo"},
+                    {mfa,         {rabbit_registry, register,
+                                   [auth_mechanism, <<"RABBIT-CR-DEMO">>,
+                                    ?MODULE]}},
+                    {requires,    rabbit_registry},
+                    {enables,     kernel_ready}]}).
+
+-record(state, {username = undefined}).
+
+%% Provides equivalent security to PLAIN but demos use of Connection.Secure(Ok)
+%% START-OK: Username
+%% SECURE: "Please tell me your password"
+%% SECURE-OK: "My password is ~s", [Password]
+
+description() ->
+    [{name, <<"RABBIT-CR-DEMO">>},
+     {description, <<"RabbitMQ Demo challenge-response authentication "
+                     "mechanism">>}].
+
+init(_Sock) ->
+    #state{}.
+
+handle_response(Response, State = #state{username = undefined}) ->
+    {challenge, <<"Please tell me your password">>,
+     State#state{username = Response}};
+
+handle_response(Response, #state{username = Username}) ->
+    case Response of
+        <<"My password is ", Password/binary>> ->
+            rabbit_access_control:check_user_pass_login(Username, Password);
+        _ ->
+            {protocol_error, "Invalid response '~s'", [Response]}
+    end.
diff --git a/src/rabbit_auth_mechanism_external.erl b/src/rabbit_auth_mechanism_external.erl
new file mode 100644
index 00000000..b21dd313
--- /dev/null
+++ b/src/rabbit_auth_mechanism_external.erl
@@ -0,0 +1,107 @@
+%%   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 Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_auth_mechanism_external).
+-include("rabbit.hrl").
+
+-behaviour(rabbit_auth_mechanism).
+
+-export([description/0, init/1, handle_response/2]).
+
+-include("rabbit_auth_mechanism_spec.hrl").
+
+-include_lib("public_key/include/public_key.hrl").
+
+-rabbit_boot_step({?MODULE,
+                   [{description, "auth mechanism external"},
+                    {mfa,         {rabbit_registry, register,
+                                   [auth_mechanism, <<"EXTERNAL">>, ?MODULE]}},
+                    {requires,    rabbit_registry},
+                    {enables,     kernel_ready}]}).
+
+-record(state, {username = undefined}).
+
+%% SASL EXTERNAL. SASL says EXTERNAL means "use credentials
+%% established by means external to the mechanism". We define that to
+%% mean the peer certificate's subject's CN.
+
+description() ->
+    [{name, <<"EXTERNAL">>},
+     {description, <<"SASL EXTERNAL authentication mechanism">>}].
+
+init(Sock) ->
+    Username = case rabbit_net:peercert(Sock) of
+                   {ok, C} ->
+                       CN = case rabbit_ssl:peer_cert_subject_item(
+                                   C, ?'id-at-commonName') of
+                                not_found -> {refused, "no CN found"};
+                                CN0       -> list_to_binary(CN0)
+                            end,
+                       case config_sane() of
+                           true  -> CN;
+                           false -> {refused, "configuration unsafe"}
+                       end;
+                   {error, no_peercert} ->
+                       {refused, "no peer certificate"};
+                   nossl ->
+                       {refused, "not SSL connection"}
+               end,
+    #state{username = Username}.
+
+handle_response(_Response, #state{username = Username}) ->
+    case Username of
+        {refused, _} = E ->
+            E;
+        _ ->
+            case rabbit_access_control:lookup_user(Username) of
+                {ok, User} ->
+                    {ok, User};
+                {error, not_found} ->
+                    %% This is not an information leak as we have to
+                    %% have validated a client cert to get this far.
+                    {refused, io_lib:format("user '~s' not found", [Username])}
+            end
+    end.
+
+%%--------------------------------------------------------------------------
+
+config_sane() ->
+    {ok, Opts} = application:get_env(ssl_options),
+    case {proplists:get_value(fail_if_no_peer_cert, Opts),
+          proplists:get_value(verify, Opts)} of
+        {true, verify_peer} ->
+            true;
+        {F, V} ->
+            rabbit_log:warning("EXTERNAL mechanism disabled, "
+                               "fail_if_no_peer_cert=~p; "
+                               "verify=~p~n", [F, V]),
+            false
+    end.
diff --git a/src/rabbit_auth_mechanism_plain.erl b/src/rabbit_auth_mechanism_plain.erl
new file mode 100644
index 00000000..4171b999
--- /dev/null
+++ b/src/rabbit_auth_mechanism_plain.erl
@@ -0,0 +1,65 @@
+%%   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 Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2010 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_auth_mechanism_plain).
+-include("rabbit.hrl").
+
+-behaviour(rabbit_auth_mechanism).
+
+-export([description/0, init/1, handle_response/2]).
+
+-include("rabbit_auth_mechanism_spec.hrl").
+
+-rabbit_boot_step({?MODULE,
+                   [{description, "auth mechanism plain"},
+                    {mfa,         {rabbit_registry, register,
+                                   [auth_mechanism, <<"PLAIN">>, ?MODULE]}},
+                    {requires,    rabbit_registry},
+                    {enables,     kernel_ready}]}).
+
+%% SASL PLAIN, as used by the Qpid Java client and our clients. Also,
+%% apparently, by OpenAMQ.
+
+description() ->
+    [{name, <<"PLAIN">>},
+     {description, <<"SASL PLAIN authentication mechanism">>}].
+
+init(_Sock) ->
+    [].
+
+handle_response(Response, _State) ->
+    case re:run(Response, "\\0([^\\0]*)", [{capture, all_but_first, binary},
+                                           global]) of
+        {match, [[User],[Pass]]} ->
+            rabbit_access_control:check_user_pass_login(User, Pass);
+        _ ->
+            {refused, io_lib:format("response ~p invalid", [Response])}
+    end.
diff --git a/src/rabbit_binary_generator.erl b/src/rabbit_binary_generator.erl
index b2997ae2..a5297a70 100644
--- a/src/rabbit_binary_generator.erl
+++ b/src/rabbit_binary_generator.erl
@@ -49,8 +49,6 @@
 -export([ensure_content_encoded/2, clear_encoded_content/1]).
 -export([map_exception/3]).
 
--import(lists).
-
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
diff --git a/src/rabbit_binary_parser.erl b/src/rabbit_binary_parser.erl
index ebf063f0..4b4358b4 100644
--- a/src/rabbit_binary_parser.erl
+++ b/src/rabbit_binary_parser.erl
@@ -36,8 +36,6 @@
 -export([parse_table/1, parse_properties/2]).
 -export([ensure_content_decoded/1, clear_decoded_content/1]).
 
--import(lists).
-
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
diff --git a/src/rabbit_binding.erl b/src/rabbit_binding.erl
index 668fb9bb..ccadf5af 100644
--- a/src/rabbit_binding.erl
+++ b/src/rabbit_binding.erl
@@ -305,7 +305,7 @@ table_for_resource(#resource{kind = queue})    -> rabbit_queue.
 
 %% Used with atoms from records; e.g., the type is expected to exist.
 type_to_module(T) ->
-    {ok, Module} = rabbit_exchange_type_registry:lookup_module(T),
+    {ok, Module} = rabbit_registry:lookup_module(exchange, T),
     Module.
 
 contains(Table, MatchHead) ->
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index f7b875a0..4e9bd4b1 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -35,10 +35,10 @@
 
 -behaviour(gen_server2).
 
--export([start_link/7, do/2, do/3, shutdown/1]).
--export([send_command/2, deliver/4, flushed/2]).
+-export([start_link/7, do/2, do/3, flush/1, shutdown/1]).
+-export([send_command/2, deliver/4, flushed/2, confirm/2, flush_confirms/1]).
 -export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1]).
--export([emit_stats/1, flush/1, flush_multiple_acks/1, confirm/2]).
+-export([emit_stats/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,
@@ -49,7 +49,7 @@
              uncommitted_ack_q, unacked_message_q,
              username, virtual_host, most_recently_declared_queue,
              consumer_mapping, blocking, queue_collector_pid, stats_timer,
-             confirm_enabled, published_count, confirm_multiple, confirm_tref,
+             confirm_enabled, publish_seqno, confirm_multiple, confirm_tref,
              held_confirms, unconfirmed, queues_for_msg}).
 
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
@@ -72,7 +72,7 @@
 
 -define(INFO_KEYS, ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [pid]).
 
--define(FLUSH_MULTIPLE_ACKS_INTERVAL, 1000).
+-define(FLUSH_CONFIRMS_INTERVAL, 1000).
 
 %%----------------------------------------------------------------------------
 
@@ -90,12 +90,15 @@
 -spec(do/2 :: (pid(), rabbit_framing:amqp_method_record()) -> 'ok').
 -spec(do/3 :: (pid(), rabbit_framing:amqp_method_record(),
                rabbit_types:maybe(rabbit_types:content())) -> 'ok').
+-spec(flush/1 :: (pid()) -> 'ok').
 -spec(shutdown/1 :: (pid()) -> 'ok').
 -spec(send_command/2 :: (pid(), rabbit_framing:amqp_method_record()) -> 'ok').
 -spec(deliver/4 ::
         (pid(), rabbit_types:ctag(), boolean(), rabbit_amqqueue:qmsg())
         -> 'ok').
 -spec(flushed/2 :: (pid(), pid()) -> 'ok').
+-spec(confirm/2 ::(pid(), non_neg_integer()) -> 'ok').
+-spec(flush_confirms/1 :: (pid()) -> 'ok').
 -spec(list/0 :: () -> [pid()]).
 -spec(info_keys/0 :: () -> rabbit_types:info_keys()).
 -spec(info/1 :: (pid()) -> rabbit_types:infos()).
@@ -103,8 +106,6 @@
 -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(flush_multiple_acks/1 :: (pid()) -> 'ok').
--spec(confirm/2 ::(pid(), non_neg_integer()) -> 'ok').
 
 -endif.
 
@@ -121,6 +122,9 @@ do(Pid, Method) ->
 do(Pid, Method, Content) ->
     gen_server2:cast(Pid, {method, Method, Content}).
 
+flush(Pid) ->
+    gen_server2:call(Pid, flush).
+
 shutdown(Pid) ->
     gen_server2:cast(Pid, terminate).
 
@@ -133,6 +137,12 @@ deliver(Pid, ConsumerTag, AckRequired, Msg) ->
 flushed(Pid, QPid) ->
     gen_server2:cast(Pid, {flushed, QPid}).
 
+confirm(Pid, MsgSeqNo) ->
+    gen_server2:cast(Pid, {confirm, MsgSeqNo, self()}).
+
+flush_confirms(Pid) ->
+    gen_server2:cast(Pid, flush_confirms).
+
 list() ->
     pg_local:get_members(rabbit_channels).
 
@@ -156,15 +166,6 @@ info_all(Items) ->
 emit_stats(Pid) ->
     gen_server2:cast(Pid, emit_stats).
 
-flush(Pid) ->
-    gen_server2:call(Pid, flush).
-
-flush_multiple_acks(Pid) ->
-    gen_server2:cast(Pid, flush_multiple_acks).
-
-confirm(Pid, MsgSeqNo) ->
-    gen_server2:cast(Pid, {confirm, MsgSeqNo, self()}).
-
 %%---------------------------------------------------------------------------
 
 init([Channel, ReaderPid, WriterPid, Username, VHost, CollectorPid,
@@ -191,7 +192,7 @@ init([Channel, ReaderPid, WriterPid, Username, VHost, CollectorPid,
                 queue_collector_pid     = CollectorPid,
                 stats_timer             = StatsTimer,
                 confirm_enabled         = false,
-                published_count         = 0,
+                publish_seqno           = 0,
                 confirm_multiple        = false,
                 held_confirms           = gb_sets:new(),
                 unconfirmed             = gb_sets:new(),
@@ -215,6 +216,9 @@ prioritise_cast(Msg, _State) ->
         _          -> 0
     end.
 
+handle_call(flush, _From, State) ->
+    reply(ok, State);
+
 handle_call(info, _From, State) ->
     reply(infos(?INFO_KEYS, State), State);
 
@@ -224,9 +228,6 @@ handle_call({info, Items}, _From, State) ->
     catch Error -> reply({error, Error}, State)
     end;
 
-handle_call(flush, _From, State) ->
-    reply(ok, State);
-
 handle_call(_Request, _From, State) ->
     noreply(State).
 
@@ -260,14 +261,24 @@ handle_cast({command, Msg}, State = #ch{writer_pid = WriterPid}) ->
     ok = rabbit_writer:send_command(WriterPid, Msg),
     noreply(State);
 
-handle_cast({deliver, ConsumerTag, AckRequired, Msg},
+handle_cast({deliver, ConsumerTag, AckRequired,
+             Msg = {_QName, QPid, _MsgId, Redelivered,
+                    #basic_message{exchange_name = ExchangeName,
+                                   routing_key = RoutingKey,
+                                   content = Content}}},
             State = #ch{writer_pid = WriterPid,
                         next_tag = DeliveryTag}) ->
     State1 = lock_message(AckRequired,
                           ack_record(DeliveryTag, ConsumerTag, Msg),
                           State),
-    ok = internal_deliver(WriterPid, true, ConsumerTag, DeliveryTag, Msg),
-    {_QName, QPid, _MsgId, _Redelivered, _Msg} = Msg,
+
+    M = #'basic.deliver'{consumer_tag = ConsumerTag,
+                         delivery_tag = DeliveryTag,
+                         redelivered = Redelivered,
+                         exchange = ExchangeName#resource.name,
+                         routing_key = RoutingKey},
+    rabbit_writer:send_command_and_notify(WriterPid, QPid, self(), M, Content),
+
     maybe_incr_stats([{QPid, 1}],
                      case AckRequired of
                          true  -> deliver;
@@ -281,11 +292,11 @@ handle_cast(emit_stats, State = #ch{stats_timer = StatsTimer}) ->
      State#ch{stats_timer = rabbit_event:reset_stats_timer(StatsTimer)},
      hibernate};
 
-handle_cast(flush_multiple_acks, State) ->
-    {noreply, flush_multiple(State)};
+handle_cast(flush_confirms, State) ->
+    {noreply, internal_flush_confirms(State)};
 
 handle_cast({confirm, MsgSeqNo, From}, State) ->
-    {noreply, send_or_enqueue_ack(MsgSeqNo, From, State)}.
+    {noreply, confirm(MsgSeqNo, From, State)}.
 
 handle_info({'DOWN', _MRef, process, QPid, _Reason},
             State = #ch{queues_for_msg = QFM}) ->
@@ -293,7 +304,7 @@ handle_info({'DOWN', _MRef, process, QPid, _Reason},
                fun(Msg, QPids, State0 = #ch{queues_for_msg = QFM0}) ->
                        Qs = sets:del_element(QPid, QPids),
                        case sets:size(Qs) of
-                           0 -> send_or_enqueue_ack(Msg, QPid, State0);
+                           0 -> confirm(Msg, QPid, State0);
                            _ -> State0#ch{queues_for_msg =
                                               dict:store(Msg, Qs, QFM0)}
                        end
@@ -303,7 +314,7 @@ handle_info({'DOWN', _MRef, process, QPid, _Reason},
 
 handle_pre_hibernate(State = #ch{stats_timer   = StatsTimer}) ->
     ok = clear_permission_cache(),
-    State1 = flush_multiple(State),
+    State1 = internal_flush_confirms(State),
     rabbit_event:if_enabled(StatsTimer,
                             fun () ->
                                     internal_emit_stats(
@@ -455,25 +466,24 @@ queue_blocked(QPid, State = #ch{blocking = Blocking}) ->
                       State#ch{blocking = Blocking1}
     end.
 
-send_or_enqueue_ack(undefined, _QPid, State) ->
+confirm(undefined, _QPid, State) ->
     State;
-send_or_enqueue_ack(_MsgSeqNo, _QPid, State = #ch{confirm_enabled = false}) ->
+confirm(_MsgSeqNo, _QPid, State = #ch{confirm_enabled = false}) ->
     State;
-send_or_enqueue_ack(MsgSeqNo, QPid, State = #ch{confirm_multiple = false}) ->
-    do_if_unconfirmed(
-      MsgSeqNo, QPid,
-      fun(MSN, State1 = #ch{writer_pid = WriterPid}) ->
-              ok = rabbit_writer:send_command(
-                     WriterPid, #'basic.ack'{delivery_tag = MSN}),
-              State1
-      end, State);
-send_or_enqueue_ack(MsgSeqNo, QPid, State = #ch{confirm_multiple = true}) ->
-    do_if_unconfirmed(
-      MsgSeqNo, QPid,
-      fun(MSN, State1 = #ch{held_confirms = As}) ->
-              start_confirm_timer(
-                State1#ch{held_confirms = gb_sets:add(MSN, As)})
-      end, State).
+confirm(MsgSeqNo, QPid, State = #ch{confirm_multiple = false}) ->
+    do_if_unconfirmed(MsgSeqNo, QPid,
+                      fun(MSN, State1 = #ch{writer_pid = WriterPid}) ->
+                              ok = rabbit_writer:send_command(
+                                     WriterPid, #'basic.ack'{
+                                       delivery_tag = MSN}),
+                              State1
+                      end, State);
+confirm(MsgSeqNo, QPid, State = #ch{confirm_multiple = true}) ->
+    do_if_unconfirmed(MsgSeqNo, QPid,
+                      fun(MSN, State1 = #ch{held_confirms = As}) ->
+                              start_confirm_timer(
+                                State1#ch{held_confirms = gb_sets:add(MSN, As)})
+                      end, State).
 
 do_if_unconfirmed(MsgSeqNo, QPid, ConfirmFun,
                   State = #ch{unconfirmed    = UC,
@@ -484,9 +494,8 @@ do_if_unconfirmed(MsgSeqNo, QPid, ConfirmFun,
             Unconfirmed1 = gb_sets:delete(MsgSeqNo, UC),
             case QPid of
                 undefined ->
-                    ConfirmFun(MsgSeqNo,
-                               State#ch{unconfirmed = Unconfirmed1});
-                _        ->
+                    ConfirmFun(MsgSeqNo, State#ch{unconfirmed = Unconfirmed1});
+                _  ->
                     {ok, Qs} = dict:find(MsgSeqNo, QFM),
                     Qs1 = sets:del_element(QPid, Qs),
                     case sets:size(Qs1) of
@@ -499,7 +508,8 @@ do_if_unconfirmed(MsgSeqNo, QPid, ConfirmFun,
                                           dict:store(MsgSeqNo, Qs1, QFM)}
                     end
             end;
-        false -> State
+        false ->
+            State
     end.
 
 handle_method(#'channel.open'{}, _, State = #ch{state = starting}) ->
@@ -537,11 +547,11 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     {MsgSeqNo, State1}
         = case ConfirmEnabled of
               false -> {undefined, State};
-              true  -> Count = State#ch.published_count,
-                       {Count,
-                        State#ch{published_count = Count + 1,
+              true  -> SeqNo = State#ch.publish_seqno,
+                       {SeqNo,
+                        State#ch{publish_seqno = SeqNo + 1,
                                  unconfirmed =
-                                     gb_sets:add(Count, State#ch.unconfirmed)}}
+                                     gb_sets:add(SeqNo, State#ch.unconfirmed)}}
           end,
     Message = #basic_message{exchange_name = ExchangeName,
                              routing_key   = RoutingKey,
@@ -1222,12 +1232,12 @@ is_message_persistent(Content) ->
 
 process_routing_result(unroutable,    _, MsgSeqNo, Message, State) ->
     ok = basic_return(Message, State#ch.writer_pid, no_route),
-    send_or_enqueue_ack(MsgSeqNo, undefined, State);
+    confirm(MsgSeqNo, undefined, State);
 process_routing_result(not_delivered, _, MsgSeqNo, Message, State) ->
     ok = basic_return(Message, State#ch.writer_pid, no_consumers),
-    send_or_enqueue_ack(MsgSeqNo, undefined, State);
+    confirm(MsgSeqNo, undefined, State);
 process_routing_result(routed,       [], MsgSeqNo,       _, State) ->
-    send_or_enqueue_ack(MsgSeqNo, undefined, State);
+    confirm(MsgSeqNo, undefined, State);
 process_routing_result(routed,        _, undefined,      _, State) ->
     State;
 process_routing_result(routed,    QPids, MsgSeqNo,       _,
@@ -1241,21 +1251,44 @@ lock_message(true, MsgStruct, State = #ch{unacked_message_q = UAMQ}) ->
 lock_message(false, _MsgStruct, State) ->
     State.
 
-internal_deliver(WriterPid, Notify, ConsumerTag, DeliveryTag,
-                 {_QName, QPid, _MsgId, Redelivered,
-                  #basic_message{exchange_name = ExchangeName,
-                                 routing_key = RoutingKey,
-                                 content = Content}}) ->
-    M = #'basic.deliver'{consumer_tag = ConsumerTag,
-                         delivery_tag = DeliveryTag,
-                         redelivered = Redelivered,
-                         exchange = ExchangeName#resource.name,
-                         routing_key = RoutingKey},
-    ok = case Notify of
-             true  -> rabbit_writer:send_command_and_notify(
-                        WriterPid, QPid, self(), M, Content);
-             false -> rabbit_writer:send_command(WriterPid, M, Content)
-         end.
+start_confirm_timer(State = #ch{confirm_tref = undefined}) ->
+    {ok, TRef} = timer:apply_after(?FLUSH_CONFIRMS_INTERVAL,
+                                   ?MODULE, flush_confirms, [self()]),
+    State#ch{confirm_tref = TRef};
+start_confirm_timer(State) ->
+    State.
+
+stop_confirm_timer(State = #ch{confirm_tref = undefined}) ->
+    State;
+stop_confirm_timer(State = #ch{confirm_tref = TRef}) ->
+    {ok, cancel} = timer:cancel(TRef),
+    State#ch{confirm_tref = undefined}.
+
+internal_flush_confirms(State = #ch{writer_pid    = WriterPid,
+                                    held_confirms = Cs}) ->
+    case gb_sets:is_empty(Cs) of
+        true  -> State#ch{confirm_tref = undefined};
+        false -> [First | Rest] = gb_sets:to_list(Cs),
+                 {Mult, Inds} = find_consecutive_sequence(First, Rest),
+                 ok = rabbit_writer:send_command(
+                        WriterPid,
+                        #'basic.ack'{delivery_tag = Mult, multiple = true}),
+                 ok = lists:foldl(
+                        fun(T, ok) -> rabbit_writer:send_command(
+                                        WriterPid,
+                                        #'basic.ack'{delivery_tag = T})
+                        end, ok, Inds),
+                 State#ch{held_confirms = gb_sets:new(),
+                          confirm_tref  = undefined}
+    end.
+
+%% Find longest sequence of consecutive numbers at the beginning.
+find_consecutive_sequence(Last, []) ->
+    {Last, []};
+find_consecutive_sequence(Last, [N | Ns]) when N == (Last + 1) ->
+    find_consecutive_sequence(N, Ns);
+find_consecutive_sequence(Last, Ns) ->
+    {Last, Ns}.
 
 terminate(State) ->
     stop_confirm_timer(State),
@@ -1344,42 +1377,3 @@ erase_queue_stats(QPid) ->
     erase({queue_stats, QPid}),
     [erase({queue_exchange_stats, QX}) ||
         {{queue_exchange_stats, QX = {QPid0, _}}, _} <- get(), QPid =:= QPid0].
-
-start_confirm_timer(State = #ch{confirm_tref = undefined}) ->
-    {ok, TRef} = timer:apply_after(?FLUSH_MULTIPLE_ACKS_INTERVAL,
-                                   ?MODULE, flush_multiple_acks, [self()]),
-    State#ch{confirm_tref = TRef};
-start_confirm_timer(State) ->
-    State.
-
-stop_confirm_timer(State = #ch{confirm_tref = undefined}) ->
-    State;
-stop_confirm_timer(State = #ch{confirm_tref = TRef}) ->
-    {ok, cancel} = timer:cancel(TRef),
-    State#ch{confirm_tref = undefined}.
-
-flush_multiple(State = #ch{writer_pid    = WriterPid,
-                           held_confirms = Cs}) ->
-    case gb_sets:is_empty(Cs) of
-        true  -> State#ch{confirm_tref = undefined};
-        false -> [First | Rest] = gb_sets:to_list(Cs),
-                 {Mult, Inds} = find_consecutive_sequence(First, Rest),
-                 ok = rabbit_writer:send_command(
-                        WriterPid,
-                        #'basic.ack'{delivery_tag = Mult, multiple = true}),
-                 ok = lists:foldl(
-                        fun(T, ok) -> rabbit_writer:send_command(
-                                        WriterPid,
-                                        #'basic.ack'{delivery_tag = T})
-                        end, ok, Inds),
-                 State#ch{held_confirms = gb_sets:new(),
-                          confirm_tref  = undefined}
-    end.
-
-%% Find longest sequence of consecutive numbers at the beginning.
-find_consecutive_sequence(Last, []) ->
-    {Last, []};
-find_consecutive_sequence(Last, [N | Ns]) when N == (Last + 1) ->
-    find_consecutive_sequence(N, Ns);
-find_consecutive_sequence(Last, Ns) ->
-    {Last, Ns}.
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 72b77b1f..360217a2 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -32,7 +32,7 @@
 -module(rabbit_control).
 -include("rabbit.hrl").
 
--export([start/0, stop/0, action/5]).
+-export([start/0, stop/0, action/5, diagnostics/1]).
 
 -define(RPC_TIMEOUT, infinity).
 
@@ -50,6 +50,7 @@
         (atom(), node(), [string()], [{string(), any()}],
          fun ((string(), [any()]) -> 'ok'))
         -> 'ok').
+-spec(diagnostics/1 :: (node()) -> [{string(), [any()]}]).
 -spec(usage/0 :: () -> no_return()).
 
 -endif.
@@ -116,24 +117,28 @@ fmt_stderr(Format, Args) -> rabbit_misc:format_stderr(Format ++ "~n", Args).
 print_error(Format, Args) -> fmt_stderr("Error: " ++ Format, Args).
 
 print_badrpc_diagnostics(Node) ->
-    fmt_stderr("diagnostics:", []),
+    [fmt_stderr(Fmt, Args) || {Fmt, Args} <- diagnostics(Node)].
+
+diagnostics(Node) ->
     {_NodeName, NodeHost} = rabbit_misc:nodeparts(Node),
-    case net_adm:names(NodeHost) of
-        {error, EpmdReason} ->
-            fmt_stderr("- unable to connect to epmd on ~s: ~w",
-                       [NodeHost, EpmdReason]);
-        {ok, NamePorts} ->
-            fmt_stderr("- nodes and their ports on ~s: ~p",
-                       [NodeHost, [{list_to_atom(Name), Port} ||
-                                      {Name, Port} <- NamePorts]])
-            end,
-    fmt_stderr("- current node: ~w", [node()]),
-    case init:get_argument(home) of
-        {ok, [[Home]]} -> fmt_stderr("- current node home dir: ~s", [Home]);
-        Other          -> fmt_stderr("- no current node home dir: ~p", [Other])
-    end,
-    fmt_stderr("- current node cookie hash: ~s", [rabbit_misc:cookie_hash()]),
-    ok.
+    [
+        {"diagnostics:", []},
+        case net_adm:names(NodeHost) of
+            {error, EpmdReason} ->
+                {"- unable to connect to epmd on ~s: ~w",
+                    [NodeHost, EpmdReason]};
+            {ok, NamePorts} ->
+                {"- nodes and their ports on ~s: ~p",
+                              [NodeHost, [{list_to_atom(Name), Port} ||
+                                          {Name, Port} <- NamePorts]]}
+        end,
+        {"- current node: ~w", [node()]},
+        case init:get_argument(home) of
+            {ok, [[Home]]} -> {"- current node home dir: ~s", [Home]};
+            Other          -> {"- no current node home dir: ~p", [Other]}
+        end,
+        {"- current node cookie hash: ~s", [rabbit_misc:cookie_hash()]}
+    ].
 
 stop() ->
     ok.
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 00e479a2..7414c904 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -150,17 +150,17 @@ declare(XName, Type, Durable, AutoDelete, Args) ->
 
 %% Used with atoms from records; e.g., the type is expected to exist.
 type_to_module(T) ->
-    {ok, Module} = rabbit_exchange_type_registry:lookup_module(T),
+    {ok, Module} = rabbit_registry:lookup_module(exchange, T),
     Module.
 
 %% Used with binaries sent over the wire; the type may not exist.
 check_type(TypeBin) ->
-    case rabbit_exchange_type_registry:binary_to_type(TypeBin) of
+    case rabbit_registry:binary_to_type(TypeBin) of
         {error, not_found} ->
             rabbit_misc:protocol_error(
               command_invalid, "unknown exchange type '~s'", [TypeBin]);
         T ->
-            case rabbit_exchange_type_registry:lookup_module(T) of
+            case rabbit_registry:lookup_module(exchange, T) of
                 {error, not_found} -> rabbit_misc:protocol_error(
                                         command_invalid,
                                         "invalid exchange type '~s'", [T]);
diff --git a/src/rabbit_exchange_type_direct.erl b/src/rabbit_exchange_type_direct.erl
index d934a497..d49d0199 100644
--- a/src/rabbit_exchange_type_direct.erl
+++ b/src/rabbit_exchange_type_direct.erl
@@ -41,9 +41,9 @@
 
 -rabbit_boot_step({?MODULE,
                    [{description, "exchange type direct"},
-                    {mfa,         {rabbit_exchange_type_registry, register,
-                                   [<<"direct">>, ?MODULE]}},
-                    {requires,    rabbit_exchange_type_registry},
+                    {mfa,         {rabbit_registry, register,
+                                   [exchange, <<"direct">>, ?MODULE]}},
+                    {requires,    rabbit_registry},
                     {enables,     kernel_ready}]}).
 
 description() ->
diff --git a/src/rabbit_exchange_type_fanout.erl b/src/rabbit_exchange_type_fanout.erl
index 77ca9686..e7f75464 100644
--- a/src/rabbit_exchange_type_fanout.erl
+++ b/src/rabbit_exchange_type_fanout.erl
@@ -41,9 +41,9 @@
 
 -rabbit_boot_step({?MODULE,
                    [{description, "exchange type fanout"},
-                    {mfa,         {rabbit_exchange_type_registry, register,
-                                   [<<"fanout">>, ?MODULE]}},
-                    {requires,    rabbit_exchange_type_registry},
+                    {mfa,         {rabbit_registry, register,
+                                   [exchange, <<"fanout">>, ?MODULE]}},
+                    {requires,    rabbit_registry},
                     {enables,     kernel_ready}]}).
 
 description() ->
diff --git a/src/rabbit_exchange_type_headers.erl b/src/rabbit_exchange_type_headers.erl
index ec9e7ba4..caf141fe 100644
--- a/src/rabbit_exchange_type_headers.erl
+++ b/src/rabbit_exchange_type_headers.erl
@@ -42,9 +42,9 @@
 
 -rabbit_boot_step({?MODULE,
                    [{description, "exchange type headers"},
-                    {mfa,         {rabbit_exchange_type_registry, register,
-                                   [<<"headers">>, ?MODULE]}},
-                    {requires,    rabbit_exchange_type_registry},
+                    {mfa,         {rabbit_registry, register,
+                                   [exchange, <<"headers">>, ?MODULE]}},
+                    {requires,    rabbit_registry},
                     {enables,     kernel_ready}]}).
 
 -ifdef(use_specs).
diff --git a/src/rabbit_exchange_type_topic.erl b/src/rabbit_exchange_type_topic.erl
index d3ecdd4d..44851858 100644
--- a/src/rabbit_exchange_type_topic.erl
+++ b/src/rabbit_exchange_type_topic.erl
@@ -41,9 +41,9 @@
 
 -rabbit_boot_step({?MODULE,
                    [{description, "exchange type topic"},
-                    {mfa,         {rabbit_exchange_type_registry, register,
-                                   [<<"topic">>, ?MODULE]}},
-                    {requires,    rabbit_exchange_type_registry},
+                    {mfa,         {rabbit_registry, register,
+                                   [exchange, <<"topic">>, ?MODULE]}},
+                    {requires,    rabbit_registry},
                     {enables,     kernel_ready}]}).
 
 -export([topic_matches/2]).
diff --git a/src/rabbit_invariable_queue.erl b/src/rabbit_invariable_queue.erl
deleted file mode 100644
index 5a0532ea..00000000
--- a/src/rabbit_invariable_queue.erl
+++ /dev/null
@@ -1,314 +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 Developers of the Original Code are LShift Ltd,
-%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
-%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
-%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
-%%   Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
-%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
-%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
-%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
-%%   (C) 2007-2010 Rabbit Technologies Ltd.
-%%
-%%   All Rights Reserved.
-%%
-%%   Contributor(s): ______________________________________.
-%%
-
--module(rabbit_invariable_queue).
-
--export([init/3, terminate/1, delete_and_terminate/1, purge/1, publish/3,
-         publish_delivered/4, fetch/2, ack/2, tx_publish/4, tx_ack/3,
-         dropwhile/2, tx_rollback/2, tx_commit/4, requeue/3, len/1, is_empty/1,
-         set_ram_duration_target/2, ram_duration/1, needs_idle_timeout/1,
-         idle_timeout/1, handle_pre_hibernate/1, status/1]).
-
--export([start/1, stop/0]).
-
--behaviour(rabbit_backing_queue).
-
--include("rabbit.hrl").
-
--record(iv_state, { queue, qname, durable, len, pending_ack }).
--record(tx, { pending_messages, pending_acks, is_persistent }).
-
--ifdef(use_specs).
-
--type(ack() :: rabbit_guid:guid() | 'blank_ack').
--type(state() :: #iv_state { queue       :: queue(),
-                             qname       :: rabbit_amqqueue:name(),
-                             len         :: non_neg_integer(),
-                             pending_ack :: dict()
-                           }).
--include("rabbit_backing_queue_spec.hrl").
-
--endif.
-
-start(DurableQueues) ->
-    ok = rabbit_sup:start_child(rabbit_persister, [DurableQueues]).
-
-stop() ->
-    ok = rabbit_sup:stop_child(rabbit_persister).
-
-init(QName, IsDurable, Recover) ->
-    Q = queue:from_list(case IsDurable andalso Recover of
-                            true  -> rabbit_persister:queue_content(QName);
-                            false -> []
-                        end),
-    #iv_state { queue       = Q,
-                qname       = QName,
-                durable     = IsDurable,
-                len         = queue:len(Q),
-                pending_ack = dict:new() }.
-
-terminate(State) ->
-    State #iv_state { queue = queue:new(), len = 0, pending_ack = dict:new() }.
-
-delete_and_terminate(State = #iv_state { qname = QName, durable = IsDurable,
-                                         pending_ack = PA }) ->
-    ok = persist_acks(QName, IsDurable, none, dict:fetch_keys(PA), PA),
-    {_PLen, State1} = purge(State),
-    terminate(State1).
-
-purge(State = #iv_state { queue = Q, qname = QName, durable = IsDurable,
-                          len = Len }) ->
-    %% We do not purge messages pending acks.
-    {AckTags, PA} =
-        rabbit_misc:queue_fold(
-          fun ({#basic_message { is_persistent = false },
-                _MsgProps, _IsDelivered}, Acc) ->
-                  Acc;
-              ({Msg = #basic_message { guid = Guid }, MsgProps, IsDelivered},
-               {AckTagsN, PAN}) ->
-                  ok = persist_delivery(QName, IsDurable, IsDelivered, Msg),
-                  {[Guid | AckTagsN], store_ack(Msg, MsgProps, PAN)}
-          end, {[], dict:new()}, Q),
-    ok = persist_acks(QName, IsDurable, none, AckTags, PA),
-    {Len, State #iv_state { len = 0, queue = queue:new() }}.
-
-publish(Msg, MsgProps, State = #iv_state { queue   = Q,
-                                           qname   = QName,
-                                           durable = IsDurable,
-                                           len     = Len }) ->
-    ok = persist_message(QName, IsDurable, none, Msg, MsgProps),
-    State #iv_state { queue = enqueue(Msg, MsgProps, false, Q), len = Len + 1 }.
-
-publish_delivered(false, _Msg, _MsgProps, State) ->
-    {blank_ack, State};
-publish_delivered(true, Msg = #basic_message { guid = Guid },
-                  MsgProps,
-                  State = #iv_state { qname = QName, durable = IsDurable,
-                                      len = 0, pending_ack = PA }) ->
-    ok = persist_message(QName, IsDurable, none, Msg, MsgProps),
-    ok = persist_delivery(QName, IsDurable, false, Msg),
-    {Guid, State #iv_state { pending_ack = store_ack(Msg, MsgProps, PA) }}.
-
-dropwhile(_Pred, State = #iv_state { len = 0 }) ->
-    State;
-dropwhile(Pred, State = #iv_state { queue = Q }) ->
-    {{value, {Msg, MsgProps, IsDelivered}}, Q1} = queue:out(Q),
-    case Pred(MsgProps) of
-        true  -> {_, State1} = fetch_internal(false, Q1, Msg, MsgProps,
-                                              IsDelivered, State),
-                 dropwhile(Pred, State1);
-        false -> State
-     end.
-
-fetch(_AckRequired, State = #iv_state { len = 0 }) ->
-    {empty, State};
-fetch(AckRequired, State = #iv_state { queue = Q }) ->
-    {{value, {Msg, MsgProps, IsDelivered}}, Q1} = queue:out(Q),
-    fetch_internal(AckRequired, Q1, Msg, MsgProps, IsDelivered, State).
-
-fetch_internal(AckRequired, Q1,
-               Msg = #basic_message { guid = Guid },
-               MsgProps, IsDelivered,
-               State = #iv_state { len         = Len,
-                                   qname       = QName,
-                                   durable     = IsDurable,
-                                   pending_ack = PA }) ->
-    Len1 = Len - 1,
-    ok = persist_delivery(QName, IsDurable, IsDelivered, Msg),
-    PA1 = store_ack(Msg, MsgProps, PA),
-    {AckTag, PA2} = case AckRequired of
-                        true  -> {Guid, PA1};
-                        false -> ok = persist_acks(QName, IsDurable, none,
-                                                   [Guid], PA1),
-                                 {blank_ack, PA}
-                    end,
-    {{Msg, IsDelivered, AckTag, Len1},
-     State #iv_state { queue = Q1, len = Len1, pending_ack = PA2 }}.
-
-ack(AckTags, State = #iv_state { qname = QName, durable = IsDurable,
-                                 pending_ack = PA }) ->
-    ok = persist_acks(QName, IsDurable, none, AckTags, PA),
-    PA1 = remove_acks(AckTags, PA),
-    State #iv_state { pending_ack = PA1 }.
-
-tx_publish(Txn, Msg, MsgProps, State = #iv_state { qname   = QName,
-                                                   durable = IsDurable }) ->
-    Tx = #tx { pending_messages = Pubs } = lookup_tx(Txn),
-    store_tx(Txn, Tx #tx { pending_messages = [{Msg, MsgProps} | Pubs] }),
-    ok = persist_message(QName, IsDurable, Txn, Msg, MsgProps),
-    State.
-
-tx_ack(Txn, AckTags, State = #iv_state { qname = QName, durable = IsDurable,
-                                         pending_ack = PA }) ->
-    Tx = #tx { pending_acks = Acks } = lookup_tx(Txn),
-    store_tx(Txn, Tx #tx { pending_acks = [AckTags | Acks] }),
-    ok = persist_acks(QName, IsDurable, Txn, AckTags, PA),
-    State.
-
-tx_rollback(Txn, State = #iv_state { qname = QName }) ->
-    #tx { pending_acks = AckTags } = lookup_tx(Txn),
-    ok = do_if_persistent(fun rabbit_persister:rollback_transaction/1,
-                          Txn, QName),
-    erase_tx(Txn),
-    {lists:flatten(AckTags), State}.
-
-tx_commit(Txn, Fun, MsgPropsFun, State = #iv_state { qname       = QName,
-                                                     pending_ack = PA,
-                                                     queue       = Q,
-                                                     len         = Len }) ->
-    #tx { pending_acks = AckTags, pending_messages = PubsRev } = lookup_tx(Txn),
-    ok = do_if_persistent(fun rabbit_persister:commit_transaction/1,
-                          Txn, QName),
-    erase_tx(Txn),
-    Fun(),
-    AckTags1 = lists:flatten(AckTags),
-    PA1 = remove_acks(AckTags1, PA),
-    {Q1, Len1} = lists:foldr(fun ({Msg, MsgProps}, {QN, LenN}) ->
-                                     {enqueue(Msg, MsgPropsFun(MsgProps),
-                                              false, QN),
-                                      LenN + 1}
-                             end, {Q, Len}, PubsRev),
-    {AckTags1, State #iv_state { pending_ack = PA1, queue = Q1, len = Len1 }}.
-
-requeue(AckTags, MsgPropsFun, State = #iv_state { pending_ack = PA,
-                                                  queue       = Q,
-                                                  len         = Len }) ->
-    %% We don't need to touch the persister here - the persister will
-    %% already have these messages published and delivered as
-    %% necessary. The complication is that the persister's seq_id will
-    %% now be wrong, given the position of these messages in our queue
-    %% here. However, the persister's seq_id is only used for sorting
-    %% on startup, and requeue is silent as to where the requeued
-    %% messages should appear, thus the persister is permitted to sort
-    %% based on seq_id, even though it'll likely give a different
-    %% order to the last known state of our queue, prior to shutdown.
-    {Q1, Len1} = lists:foldl(
-                   fun (Guid, {QN, LenN}) ->
-                           {Msg = #basic_message {}, MsgProps}
-                               = dict:fetch(Guid, PA),
-                           {enqueue(Msg, MsgPropsFun(MsgProps), true, QN),
-                            LenN + 1}
-                   end, {Q, Len}, AckTags),
-    PA1 = remove_acks(AckTags, PA),
-    State #iv_state { pending_ack = PA1, queue = Q1, len = Len1 }.
-
-enqueue(Msg, MsgProps, IsDelivered, Q) ->
-    queue:in({Msg, MsgProps, IsDelivered}, Q).
-
-len(#iv_state { len = Len }) -> Len.
-
-is_empty(State) -> 0 == len(State).
-
-set_ram_duration_target(_DurationTarget, State) -> State.
-
-ram_duration(State) -> {0, State}.
-
-needs_idle_timeout(_State) -> false.
-
-idle_timeout(State) -> State.
-
-handle_pre_hibernate(State) -> State.
-
-status(_State) -> [].
-
-%%----------------------------------------------------------------------------
-
-remove_acks(AckTags, PA) -> lists:foldl(fun dict:erase/2, PA, AckTags).
-
-store_ack(Msg = #basic_message { guid = Guid }, MsgProps, PA) ->
-    dict:store(Guid, {Msg, MsgProps}, PA).
-
-%%----------------------------------------------------------------------------
-
-lookup_tx(Txn) ->
-    case get({txn, Txn}) of
-        undefined -> #tx { pending_messages = [],
-                           pending_acks     = [],
-                           is_persistent    = false };
-        V         -> V
-    end.
-
-store_tx(Txn, Tx) ->
-    put({txn, Txn}, Tx).
-
-erase_tx(Txn) ->
-    erase({txn, Txn}).
-
-mark_tx_persistent(Txn) ->
-    store_tx(Txn, (lookup_tx(Txn)) #tx { is_persistent = true }).
-
-is_tx_persistent(Txn) ->
-    (lookup_tx(Txn)) #tx.is_persistent.
-
-do_if_persistent(F, Txn, QName) ->
-    ok = case is_tx_persistent(Txn) of
-             false -> ok;
-             true  -> F({Txn, QName})
-         end.
-
-%%----------------------------------------------------------------------------
-
-persist_message(QName, true, Txn, Msg = #basic_message {
-                                    is_persistent = true }, MsgProps) ->
-    Msg1 = Msg #basic_message {
-             %% don't persist any recoverable decoded properties
-             content = rabbit_binary_parser:clear_decoded_content(
-                         Msg #basic_message.content)},
-    persist_work(Txn, QName,
-                 [{publish, Msg1, MsgProps,
-                   {QName, Msg1 #basic_message.guid}}]);
-persist_message(_QName, _IsDurable, _Txn, _Msg, _MsgProps) ->
-    ok.
-
-persist_delivery(QName, true, false, #basic_message { is_persistent = true,
-                                                      guid = Guid }) ->
-    persist_work(none, QName, [{deliver, {QName, Guid}}]);
-persist_delivery(_QName, _IsDurable, _IsDelivered, _Msg) ->
-    ok.
-
-persist_acks(QName, true, Txn, AckTags, PA) ->
-    persist_work(Txn, QName,
-                 [{ack, {QName, Guid}} || Guid <- AckTags,
-                                          begin
-                                              {Msg, _MsgProps}
-                                                  = dict:fetch(Guid, PA),
-                                              Msg #basic_message.is_persistent
-                                          end]);
-persist_acks(_QName, _IsDurable, _Txn, _AckTags, _PA) ->
-    ok.
-
-persist_work(_Txn,_QName, []) ->
-    ok;
-persist_work(none, _QName, WorkList) ->
-    rabbit_persister:dirty_work(WorkList);
-persist_work(Txn, QName, WorkList) ->
-    mark_tx_persistent(Txn),
-    rabbit_persister:extend_transaction({Txn, QName}, WorkList).
diff --git a/src/rabbit_log.erl b/src/rabbit_log.erl
index 863f77e7..a1a8364c 100644
--- a/src/rabbit_log.erl
+++ b/src/rabbit_log.erl
@@ -41,9 +41,6 @@
 -export([debug/1, debug/2, message/4, info/1, info/2,
          warning/1, warning/2, error/1, error/2]).
 
--import(io).
--import(error_logger).
-
 -define(SERVER, ?MODULE).
 
 %%----------------------------------------------------------------------------
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 1a05b729..52d76ac4 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -68,11 +68,6 @@
 -export([now_ms/0]).
 -export([lock_file/1]).
 
--import(mnesia).
--import(lists).
--import(cover).
--import(disk_log).
-
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index a62e7a6f..dadfc16e 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -388,7 +388,7 @@ init_db(ClusterNodes, Force) ->
                     ensure_version_ok(rabbit_upgrade:read_version()),
                     ensure_schema_ok();
                 {[], false, _} ->
-                    %% First RAM node in cluster, start from scratch
+                    %% Nothing there at all, start from scratch
                     ok = create_schema();
                 {[AnotherNode|_], _, _} ->
                     %% Subsequent node in cluster, catch up
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index fea7d4a8..e8b4e8e2 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -144,13 +144,13 @@
 -type(startup_fun_state() ::
         {(fun ((A) -> 'finished' | {rabbit_guid:guid(), non_neg_integer(), A})),
          A}).
--type(guid_fun() :: fun ((gb_set()) -> any())).
+-type(maybe_guid_fun() :: 'undefined' | fun ((gb_set()) -> any())).
 
 -spec(start_link/4 ::
         (atom(), file:filename(), [binary()] | 'undefined',
          startup_fun_state()) -> rabbit_types:ok_pid_or_error()).
 -spec(successfully_recovered_state/1 :: (server()) -> boolean()).
--spec(client_init/3 :: (server(), client_ref(), guid_fun()) ->
+-spec(client_init/3 :: (server(), client_ref(), maybe_guid_fun()) ->
              client_msstate()).
 -spec(client_terminate/1 :: (client_msstate()) -> 'ok').
 -spec(client_delete_and_terminate/1 :: (client_msstate()) -> 'ok').
diff --git a/src/rabbit_persister.erl b/src/rabbit_persister.erl
deleted file mode 100644
index 11056c8e..00000000
--- a/src/rabbit_persister.erl
+++ /dev/null
@@ -1,496 +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 Developers of the Original Code are LShift Ltd,
-%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
-%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
-%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
-%%   Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
-%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
-%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
-%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
-%%   (C) 2007-2010 Rabbit Technologies Ltd.
-%%
-%%   All Rights Reserved.
-%%
-%%   Contributor(s): ______________________________________.
-%%
-
--module(rabbit_persister).
-
--behaviour(gen_server).
-
--export([start_link/1]).
-
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
-         terminate/2, code_change/3]).
-
--export([transaction/1, extend_transaction/2, dirty_work/1,
-         commit_transaction/1, rollback_transaction/1,
-         force_snapshot/0, queue_content/1]).
-
--include("rabbit.hrl").
-
--define(SERVER, ?MODULE).
-
--define(LOG_BUNDLE_DELAY, 5).
--define(COMPLETE_BUNDLE_DELAY, 2).
-
--define(PERSISTER_LOG_FORMAT_VERSION, {2, 6}).
-
--record(pstate, {log_handle, entry_count, deadline,
-                 pending_logs, pending_replies, snapshot}).
-
-%% two tables for efficient persistency
-%% one maps a key to a message
-%% the other maps a key to one or more queues.
-%% The aim is to reduce the overload of storing a message multiple times
-%% when it appears in several queues.
--record(psnapshot, {transactions, messages, queues, next_seq_id}).
-
-%%----------------------------------------------------------------------------
-
--ifdef(use_specs).
-
--type(pkey() :: rabbit_guid:guid()).
--type(pmsg() :: {rabbit_amqqueue:name(), pkey()}).
-
--type(work_item() ::
-      {publish,
-       rabbit_types:message(), rabbit_types:message_properties(), pmsg()} |
-      {deliver, pmsg()} |
-      {ack, pmsg()}).
-
--spec(start_link/1 :: ([rabbit_amqqueue:name()]) ->
-                           rabbit_types:ok_pid_or_error()).
--spec(transaction/1 :: ([work_item()]) -> 'ok').
--spec(extend_transaction/2 ::
-        ({rabbit_types:txn(), rabbit_amqqueue:name()}, [work_item()])
-        -> 'ok').
--spec(dirty_work/1 :: ([work_item()]) -> 'ok').
--spec(commit_transaction/1 ::
-        ({rabbit_types:txn(), rabbit_amqqueue:name()}) -> 'ok').
--spec(rollback_transaction/1 ::
-        ({rabbit_types:txn(), rabbit_amqqueue:name()}) -> 'ok').
--spec(force_snapshot/0 :: () -> 'ok').
--spec(queue_content/1 ::
-        (rabbit_amqqueue:name()) -> [{rabbit_types:message(), boolean()}]).
-
--endif.
-
-%%----------------------------------------------------------------------------
-
-start_link(DurableQueues) ->
-    gen_server:start_link({local, ?SERVER}, ?MODULE, [DurableQueues], []).
-
-transaction(MessageList) ->
-    ?LOGDEBUG("transaction ~p~n", [MessageList]),
-    TxnKey = rabbit_guid:guid(),
-    gen_server:call(?SERVER, {transaction, TxnKey, MessageList}, infinity).
-
-extend_transaction(TxnKey, MessageList) ->
-    ?LOGDEBUG("extend_transaction ~p ~p~n", [TxnKey, MessageList]),
-    gen_server:cast(?SERVER, {extend_transaction, TxnKey, MessageList}).
-
-dirty_work(MessageList) ->
-    ?LOGDEBUG("dirty_work ~p~n", [MessageList]),
-    gen_server:cast(?SERVER, {dirty_work, MessageList}).
-
-commit_transaction(TxnKey) ->
-    ?LOGDEBUG("commit_transaction ~p~n", [TxnKey]),
-    gen_server:call(?SERVER, {commit_transaction, TxnKey}, infinity).
-
-rollback_transaction(TxnKey) ->
-    ?LOGDEBUG("rollback_transaction ~p~n", [TxnKey]),
-    gen_server:cast(?SERVER, {rollback_transaction, TxnKey}).
-
-force_snapshot() ->
-    gen_server:call(?SERVER, force_snapshot, infinity).
-
-queue_content(QName) ->
-    gen_server:call(?SERVER, {queue_content, QName}, infinity).
-
-%%--------------------------------------------------------------------
-
-init([DurableQueues]) ->
-    process_flag(trap_exit, true),
-    FileName = base_filename(),
-    ok = filelib:ensure_dir(FileName),
-    Snapshot = #psnapshot{transactions = dict:new(),
-                          messages     = ets:new(messages, []),
-                          queues       = ets:new(queues, [ordered_set]),
-                          next_seq_id  = 0},
-    LogHandle =
-        case disk_log:open([{name, rabbit_persister},
-                            {head, current_snapshot(Snapshot)},
-                            {file, FileName}]) of
-            {ok, LH} -> LH;
-            {repaired, LH, {recovered, Recovered}, {badbytes, Bad}} ->
-                WarningFun = if
-                                 Bad > 0 -> fun rabbit_log:warning/2;
-                                 true    -> fun rabbit_log:info/2
-                             end,
-                WarningFun("Repaired persister log - ~p recovered, ~p bad~n",
-                           [Recovered, Bad]),
-                LH
-        end,
-    {Res, NewSnapshot} =
-        internal_load_snapshot(LogHandle, DurableQueues, Snapshot),
-    case Res of
-        ok ->
-            ok = take_snapshot(LogHandle, NewSnapshot);
-        {error, Reason} ->
-            rabbit_log:error("Failed to load persister log: ~p~n", [Reason]),
-            ok = take_snapshot_and_save_old(LogHandle, NewSnapshot)
-    end,
-    State = #pstate{log_handle        = LogHandle,
-                    entry_count       = 0,
-                    deadline          = infinity,
-                    pending_logs      = [],
-                    pending_replies   = [],
-                    snapshot          = NewSnapshot},
-    {ok, State}.
-
-handle_call({transaction, Key, MessageList}, From, State) ->
-    NewState = internal_extend(Key, MessageList, State),
-    do_noreply(internal_commit(From, Key, NewState));
-handle_call({commit_transaction, TxnKey}, From, State) ->
-    do_noreply(internal_commit(From, TxnKey, State));
-handle_call(force_snapshot, _From, State) ->
-    do_reply(ok, flush(true, State));
-handle_call({queue_content, QName}, _From,
-            State = #pstate{snapshot = #psnapshot{messages = Messages,
-                                                  queues   = Queues}}) ->
-    MatchSpec= [{{{QName,'$1'}, '$2', '$3', '$4'}, [], 
-                 [{{'$4', '$1', '$2', '$3'}}]}],
-    do_reply([{ets:lookup_element(Messages, K, 2), MP, D} ||
-                 {_, K, D, MP} <- lists:sort(ets:select(Queues, MatchSpec))],
-             State);
-handle_call(_Request, _From, State) ->
-    {noreply, State}.
-
-handle_cast({rollback_transaction, TxnKey}, State) ->
-    do_noreply(internal_rollback(TxnKey, State));
-handle_cast({dirty_work, MessageList}, State) ->
-    do_noreply(internal_dirty_work(MessageList, State));
-handle_cast({extend_transaction, TxnKey, MessageList}, State) ->
-    do_noreply(internal_extend(TxnKey, MessageList, State));
-handle_cast(_Msg, State) ->
-    {noreply, State}.
-
-handle_info(timeout, State = #pstate{deadline = infinity}) ->
-    State1 = flush(true, State),
-    {noreply, State1, hibernate};
-handle_info(timeout, State) ->
-    do_noreply(flush(State));
-handle_info(_Info, State) ->
-    {noreply, State}.
-
-terminate(_Reason, State = #pstate{log_handle = LogHandle}) ->
-    flush(State),
-    disk_log:close(LogHandle),
-    ok.
-
-code_change(_OldVsn, State, _Extra) ->
-    {ok, flush(State)}.
-
-%%--------------------------------------------------------------------
-
-internal_extend(Key, MessageList, State) ->
-    log_work(fun (ML) -> {extend_transaction, Key, ML} end,
-             MessageList, State).
-
-internal_dirty_work(MessageList, State) ->
-    log_work(fun (ML) -> {dirty_work, ML} end,
-             MessageList, State).
-
-internal_commit(From, Key, State = #pstate{snapshot = Snapshot}) ->
-    Unit = {commit_transaction, Key},
-    NewSnapshot = internal_integrate1(Unit, Snapshot),
-    complete(From, Unit, State#pstate{snapshot = NewSnapshot}).
-
-internal_rollback(Key, State = #pstate{snapshot = Snapshot}) ->
-    Unit = {rollback_transaction, Key},
-    NewSnapshot = internal_integrate1(Unit, Snapshot),
-    log(State#pstate{snapshot = NewSnapshot}, Unit).
-
-complete(From, Item, State = #pstate{deadline = ExistingDeadline,
-                                     pending_logs = Logs,
-                                     pending_replies = Waiting}) ->
-    State#pstate{deadline = compute_deadline(
-                              ?COMPLETE_BUNDLE_DELAY, ExistingDeadline),
-                 pending_logs = [Item | Logs],
-                 pending_replies = [From | Waiting]}.
-
-%% This is made to limit disk usage by writing messages only once onto
-%% disk.  We keep a table associating pkeys to messages, and provided
-%% the list of messages to output is left to right, we can guarantee
-%% that pkeys will be a backreference to a message in memory when a
-%% "tied" is met.
-log_work(CreateWorkUnit, MessageList,
-         State = #pstate{
-           snapshot = Snapshot = #psnapshot{messages = Messages}}) ->
-    Unit = CreateWorkUnit(
-             rabbit_misc:map_in_order(
-               fun (M = {publish, Message, MsgProps, QK = {_QName, PKey}}) ->
-                       case ets:lookup(Messages, PKey) of
-                           [_] -> {tied, MsgProps, QK};
-                           []  -> ets:insert(Messages, {PKey, Message}),
-                                  M
-                       end;
-                  (M) -> M
-               end,
-               MessageList)),
-    NewSnapshot = internal_integrate1(Unit, Snapshot),
-    log(State#pstate{snapshot = NewSnapshot}, Unit).
-
-log(State = #pstate{deadline = ExistingDeadline, pending_logs = Logs},
-    Message) ->
-    State#pstate{deadline = compute_deadline(?LOG_BUNDLE_DELAY,
-                                             ExistingDeadline),
-                 pending_logs = [Message | Logs]}.
-
-base_filename() ->
-    rabbit_mnesia:dir() ++ "/rabbit_persister.LOG".
-
-take_snapshot(LogHandle, OldFileName, Snapshot) ->
-    ok = disk_log:sync(LogHandle),
-    %% current_snapshot is the Head (ie. first thing logged)
-    ok = disk_log:reopen(LogHandle, OldFileName, current_snapshot(Snapshot)).
-
-take_snapshot(LogHandle, Snapshot) ->
-    OldFileName = lists:flatten(base_filename() ++ ".previous"),
-    file:delete(OldFileName),
-    rabbit_log:info("Rolling persister log to ~p~n", [OldFileName]),
-    ok = take_snapshot(LogHandle, OldFileName, Snapshot).
-
-take_snapshot_and_save_old(LogHandle, Snapshot) ->
-    {MegaSecs, Secs, MicroSecs} = erlang:now(),
-    Timestamp = MegaSecs * 1000000 + Secs * 1000 + MicroSecs,
-    OldFileName = lists:flatten(io_lib:format("~s.saved.~p",
-                                              [base_filename(), Timestamp])),
-    rabbit_log:info("Saving persister log in ~p~n", [OldFileName]),
-    ok = take_snapshot(LogHandle, OldFileName, Snapshot).
-
-maybe_take_snapshot(Force, State = #pstate{entry_count = EntryCount,
-                                           log_handle = LH,
-                                           snapshot = Snapshot}) ->
-    {ok, MaxWrapEntries} = application:get_env(persister_max_wrap_entries),
-    if
-        Force orelse EntryCount >= MaxWrapEntries ->
-            ok = take_snapshot(LH, Snapshot),
-            State#pstate{entry_count = 0};
-        true ->
-            State
-    end.
-
-later_ms(DeltaMilliSec) ->
-    {MegaSec, Sec, MicroSec} = now(),
-    %% Note: not normalised. Unimportant for this application.
-    {MegaSec, Sec, MicroSec + (DeltaMilliSec * 1000)}.
-
-%% Result = B - A, more or less
-time_diff({B1, B2, B3}, {A1, A2, A3}) ->
-    (B1 - A1) * 1000000 + (B2 - A2) + (B3 - A3) / 1000000.0 .
-
-compute_deadline(TimerDelay, infinity) ->
-    later_ms(TimerDelay);
-compute_deadline(_TimerDelay, ExistingDeadline) ->
-    ExistingDeadline.
-
-compute_timeout(infinity) ->
-    {ok, HibernateAfter} = application:get_env(persister_hibernate_after),
-    HibernateAfter;
-compute_timeout(Deadline) ->
-    DeltaMilliSec = time_diff(Deadline, now()) * 1000.0,
-    if
-        DeltaMilliSec =< 1 ->
-            0;
-        true ->
-            round(DeltaMilliSec)
-    end.
-
-do_noreply(State = #pstate{deadline = Deadline}) ->
-    {noreply, State, compute_timeout(Deadline)}.
-
-do_reply(Reply, State = #pstate{deadline = Deadline}) ->
-    {reply, Reply, State, compute_timeout(Deadline)}.
-
-flush(State) -> flush(false, State).
-
-flush(ForceSnapshot, State = #pstate{pending_logs = PendingLogs,
-                                     pending_replies = Waiting,
-                                     log_handle = LogHandle}) ->
-    State1 = if PendingLogs /= [] ->
-                     disk_log:alog(LogHandle, lists:reverse(PendingLogs)),
-                     State#pstate{entry_count = State#pstate.entry_count + 1};
-                true ->
-                     State
-             end,
-    State2 = maybe_take_snapshot(ForceSnapshot, State1),
-    if Waiting /= [] ->
-            ok = disk_log:sync(LogHandle),
-            lists:foreach(fun (From) -> gen_server:reply(From, ok) end,
-                          Waiting);
-       true ->
-            ok
-    end,
-    State2#pstate{deadline = infinity,
-                  pending_logs = [],
-                  pending_replies = []}.
-
-current_snapshot(_Snapshot = #psnapshot{transactions = Ts,
-                                        messages     = Messages,
-                                        queues       = Queues,
-                                        next_seq_id  = NextSeqId}) ->
-    %% Avoid infinite growth of the table by removing messages not
-    %% bound to a queue anymore
-    PKeys = ets:foldl(fun ({{_QName, PKey}, _Delivered, 
-                            _MsgProps, _SeqId}, S) ->
-                              sets:add_element(PKey, S)
-                      end, sets:new(), Queues),
-    prune_table(Messages, fun (Key) -> sets:is_element(Key, PKeys) end),
-    InnerSnapshot = {{txns, Ts},
-                     {messages, ets:tab2list(Messages)},
-                     {queues, ets:tab2list(Queues)},
-                     {next_seq_id, NextSeqId}},
-    ?LOGDEBUG("Inner snapshot: ~p~n", [InnerSnapshot]),
-    {persist_snapshot, {vsn, ?PERSISTER_LOG_FORMAT_VERSION},
-     term_to_binary(InnerSnapshot)}.
-
-prune_table(Tab, Pred) ->
-    true = ets:safe_fixtable(Tab, true),
-    ok = prune_table(Tab, Pred, ets:first(Tab)),
-    true = ets:safe_fixtable(Tab, false).
-
-prune_table(_Tab, _Pred, '$end_of_table') -> ok;
-prune_table(Tab, Pred, Key) ->
-    case Pred(Key) of
-        true  -> ok;
-        false -> ets:delete(Tab, Key)
-    end,
-    prune_table(Tab, Pred, ets:next(Tab, Key)).
-
-internal_load_snapshot(LogHandle,
-                       DurableQueues,
-                       Snapshot = #psnapshot{messages = Messages,
-                                             queues = Queues}) ->
-    {K, [Loaded_Snapshot | Items]} = disk_log:chunk(LogHandle, start),
-    case check_version(Loaded_Snapshot) of
-        {ok, StateBin} ->
-            {{txns, Ts}, {messages, Ms}, {queues, Qs},
-             {next_seq_id, NextSeqId}} = binary_to_term(StateBin),
-            true = ets:insert(Messages, Ms),
-            true = ets:insert(Queues, Qs),
-            Snapshot1 = replay(Items, LogHandle, K,
-                               Snapshot#psnapshot{
-                                 transactions = Ts,
-                                 next_seq_id = NextSeqId}),
-            %% Remove all entries for queues that no longer exist.
-            %% Note that the 'messages' table is pruned when the next
-            %% snapshot is taken.
-            DurableQueuesSet = sets:from_list(DurableQueues),
-            prune_table(Snapshot1#psnapshot.queues,
-                        fun ({QName, _PKey}) ->
-                                sets:is_element(QName, DurableQueuesSet)
-                        end),
-            %% uncompleted transactions are discarded - this is TRTTD
-            %% since we only get into this code on node restart, so
-            %% any uncompleted transactions will have been aborted.
-            {ok, Snapshot1#psnapshot{transactions = dict:new()}};
-        {error, Reason} -> {{error, Reason}, Snapshot}
-    end.
-
-check_version({persist_snapshot, {vsn, ?PERSISTER_LOG_FORMAT_VERSION},
-               StateBin}) ->
-    {ok, StateBin};
-check_version({persist_snapshot, {vsn, Vsn}, _StateBin}) ->
-    {error, {unsupported_persister_log_format, Vsn}};
-check_version(_Other) ->
-    {error, unrecognised_persister_log_format}.
-
-replay([], LogHandle, K, Snapshot) ->
-    case disk_log:chunk(LogHandle, K) of
-        {K1, Items} ->
-            replay(Items, LogHandle, K1, Snapshot);
-        {K1, Items, Badbytes} ->
-            rabbit_log:warning("~p bad bytes recovering persister log~n",
-                               [Badbytes]),
-            replay(Items, LogHandle, K1, Snapshot);
-        eof -> Snapshot
-    end;
-replay([Item | Items], LogHandle, K, Snapshot) ->
-    NewSnapshot = internal_integrate_messages(Item, Snapshot),
-    replay(Items, LogHandle, K, NewSnapshot).
-
-internal_integrate_messages(Items, Snapshot) ->
-    lists:foldl(fun (Item, Snap) -> internal_integrate1(Item, Snap) end,
-                Snapshot, Items).
-
-internal_integrate1({extend_transaction, Key, MessageList},
-                    Snapshot = #psnapshot {transactions = Transactions}) ->
-    Snapshot#psnapshot{transactions = rabbit_misc:dict_cons(Key, MessageList,
-                                                            Transactions)};
-internal_integrate1({rollback_transaction, Key},
-                    Snapshot = #psnapshot{transactions = Transactions}) ->
-    Snapshot#psnapshot{transactions = dict:erase(Key, Transactions)};
-internal_integrate1({commit_transaction, Key},
-                    Snapshot = #psnapshot{transactions = Transactions,
-                                          messages     = Messages,
-                                          queues       = Queues,
-                                          next_seq_id  = SeqId}) ->
-    case dict:find(Key, Transactions) of
-        {ok, MessageLists} ->
-            ?LOGDEBUG("persist committing txn ~p~n", [Key]),
-            NextSeqId =
-                lists:foldr(
-                  fun (ML, SeqIdN) ->
-                          perform_work(ML, Messages, Queues, SeqIdN) end,
-                  SeqId, MessageLists),
-            Snapshot#psnapshot{transactions = dict:erase(Key, Transactions),
-                               next_seq_id = NextSeqId};
-        error ->
-            Snapshot
-    end;
-internal_integrate1({dirty_work, MessageList},
-                    Snapshot = #psnapshot{messages    = Messages,
-                                          queues      = Queues,
-                                          next_seq_id = SeqId}) ->
-    Snapshot#psnapshot{next_seq_id = perform_work(MessageList, Messages,
-                                                  Queues, SeqId)}.
-
-perform_work(MessageList, Messages, Queues, SeqId) ->
-    lists:foldl(fun (Item, NextSeqId) ->
-                        perform_work_item(Item, Messages, Queues, NextSeqId)
-                end, SeqId, MessageList).
-
-perform_work_item({publish, Message, MsgProps, QK = {_QName, PKey}},
-                  Messages, Queues, NextSeqId) ->
-    true = ets:insert(Messages, {PKey, Message}),
-    true = ets:insert(Queues, {QK, false, MsgProps, NextSeqId}),
-    NextSeqId + 1;
-
-perform_work_item({tied, MsgProps, QK}, _Messages, Queues, NextSeqId) ->
-    true = ets:insert(Queues, {QK, false, MsgProps, NextSeqId}),
-    NextSeqId + 1;
-
-perform_work_item({deliver, QK}, _Messages, Queues, NextSeqId) ->
-    true = ets:update_element(Queues, QK, {2, true}),
-    NextSeqId;
-
-perform_work_item({ack, QK}, _Messages, Queues, NextSeqId) ->
-    true = ets:delete(Queues, QK),
-    NextSeqId.
diff --git a/src/rabbit_plugin_activator.erl b/src/rabbit_prelaunch.erl
index 072f297e..fcc93e6f 100644
--- a/src/rabbit_plugin_activator.erl
+++ b/src/rabbit_prelaunch.erl
@@ -29,11 +29,12 @@
 %%   Contributor(s): ______________________________________.
 %%
 
--module(rabbit_plugin_activator).
+-module(rabbit_prelaunch).
 
 -export([start/0, stop/0]).
 
 -define(BaseApps, [rabbit]).
+-define(ERROR_CODE, 1).
 
 %%----------------------------------------------------------------------------
 %% Specs
@@ -52,7 +53,7 @@ start() ->
     io:format("Activating RabbitMQ plugins ...~n"),
 
     %% Determine our various directories
-    [PluginDir, UnpackedPluginDir] = init:get_plain_arguments(),
+    [PluginDir, UnpackedPluginDir, Node] = init:get_plain_arguments(),
     RootName = UnpackedPluginDir ++ "/rabbit",
 
     %% Unpack any .ez plugins
@@ -130,7 +131,10 @@ start() ->
     [io:format("* ~s-~s~n", [App, proplists:get_value(App, AppVersions)])
      || App <- PluginApps],
     io:nl(),
-    halt(),
+
+    ok = duplicate_node_check(Node),
+
+    terminate(0),
     ok.
 
 stop() ->
@@ -251,6 +255,36 @@ process_entry(Entry = {apply,{application,start_boot,[rabbit,permanent]}}) ->
 process_entry(Entry) ->
     [Entry].
 
+%% Check whether a node with the same name is already running
+duplicate_node_check([]) ->
+    %% Ignore running node while installing windows service
+    ok;
+duplicate_node_check(Node) ->
+    {NodeName, NodeHost} = rabbit_misc:nodeparts(rabbit_misc:makenode(Node)),
+    case net_adm:names(NodeHost) of
+        {ok, NamePorts}  ->
+            case proplists:is_defined(NodeName, NamePorts) of
+                     true -> io:format("node with name ~p "
+                                       "already running on ~p~n",
+                                       [NodeName, NodeHost]),
+                             [io:format(Fmt ++ "~n", Args) ||
+                              {Fmt, Args} <- rabbit_control:diagnostics(Node)],
+                             terminate(?ERROR_CODE);
+                     false -> ok
+            end;
+        {error, EpmdReason} -> terminate("unexpected epmd error: ~p~n",
+                                         [EpmdReason])
+    end.
+
 terminate(Fmt, Args) ->
     io:format("ERROR: " ++ Fmt ++ "~n", Args),
-    halt(1).
+    terminate(?ERROR_CODE).
+
+terminate(Status) ->
+    case os:type() of
+        {unix,  _} -> halt(Status);
+        {win32, _} -> init:stop(Status),
+                      receive
+                      after infinity -> ok
+                      end
+    end.
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index ba86b540..4fb680c7 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -56,14 +56,15 @@
 
 -record(v1, {parent, sock, connection, callback, recv_length, recv_ref,
              connection_state, queue_collector, heartbeater, stats_timer,
-             channel_sup_sup_pid, start_heartbeat_fun}).
+             channel_sup_sup_pid, start_heartbeat_fun, auth_mechanism,
+             auth_state}).
 
 -define(STATISTICS_KEYS, [pid, recv_oct, recv_cnt, send_oct, send_cnt,
                           send_pend, state, channels]).
 
 -define(CREATION_EVENT_KEYS, [pid, address, port, peer_address, peer_port, ssl,
                               peer_cert_subject, peer_cert_issuer,
-                              peer_cert_validity,
+                              peer_cert_validity, auth_mechanism,
                               protocol, user, vhost, timeout, frame_max,
                               client_properties]).
 
@@ -294,7 +295,9 @@ start_connection(Parent, ChannelSupSupPid, Collector, StartHeartbeatFun, Deb,
                             stats_timer         =
                                 rabbit_event:init_stats_timer(),
                             channel_sup_sup_pid = ChannelSupSupPid,
-                            start_heartbeat_fun = StartHeartbeatFun
+                            start_heartbeat_fun = StartHeartbeatFun,
+                            auth_mechanism      = none,
+                            auth_state          = none
                            },
                         handshake, 8))
     catch
@@ -681,11 +684,12 @@ handle_input(Callback, Data, _State) ->
 start_connection({ProtocolMajor, ProtocolMinor, _ProtocolRevision},
                  Protocol,
                  State = #v1{sock = Sock, connection = Connection}) ->
-    Start = #'connection.start'{ version_major = ProtocolMajor,
-                                 version_minor = ProtocolMinor,
-                                 server_properties = server_properties(),
-                                 mechanisms = <<"PLAIN AMQPLAIN">>,
-                                 locales = <<"en_US">> },
+    Start = #'connection.start'{
+      version_major = ProtocolMajor,
+      version_minor = ProtocolMinor,
+      server_properties = server_properties(),
+      mechanisms = auth_mechanisms_binary(),
+      locales = <<"en_US">> },
     ok = send_on_channel0(Sock, Start, Protocol),
     switch_callback(State#v1{connection = Connection#connection{
                                             timeout_sec = ?NORMAL_TIMEOUT,
@@ -733,19 +737,22 @@ handle_method0(MethodName, FieldsBin,
 handle_method0(#'connection.start_ok'{mechanism = Mechanism,
                                       response = Response,
                                       client_properties = ClientProperties},
-               State = #v1{connection_state = starting,
-                           connection = Connection =
-                               #connection{protocol = Protocol},
-                           sock = Sock}) ->
-    User = rabbit_access_control:check_login(Mechanism, Response),
-    Tune = #'connection.tune'{channel_max = 0,
-                              frame_max = ?FRAME_MAX,
-                              heartbeat = 0},
-    ok = send_on_channel0(Sock, Tune, Protocol),
-    State#v1{connection_state = tuning,
-             connection = Connection#connection{
-                            user = User,
-                            client_properties = ClientProperties}};
+               State0 = #v1{connection_state = starting,
+                            connection       = Connection,
+                            sock             = Sock}) ->
+    AuthMechanism = auth_mechanism_to_module(Mechanism),
+    State = State0#v1{auth_mechanism   = AuthMechanism,
+                      auth_state       = AuthMechanism:init(Sock),
+                      connection_state = securing,
+                      connection       =
+                          Connection#connection{
+                            client_properties = ClientProperties}},
+    auth_phase(Response, State);
+
+handle_method0(#'connection.secure_ok'{response = Response},
+               State = #v1{connection_state = securing}) ->
+    auth_phase(Response, State);
+
 handle_method0(#'connection.tune_ok'{frame_max = FrameMax,
                                      heartbeat = ClientHeartbeat},
                State = #v1{connection_state = tuning,
@@ -827,6 +834,60 @@ handle_method0(_Method, #v1{connection_state = S}) ->
 send_on_channel0(Sock, Method, Protocol) ->
     ok = rabbit_writer:internal_send_command(Sock, 0, Method, Protocol).
 
+auth_mechanism_to_module(TypeBin) ->
+    case rabbit_registry:binary_to_type(TypeBin) of
+        {error, not_found} ->
+            rabbit_misc:protocol_error(
+              command_invalid, "unknown authentication mechanism '~s'",
+              [TypeBin]);
+        T ->
+            case {lists:member(T, auth_mechanisms()),
+                  rabbit_registry:lookup_module(auth_mechanism, T)} of
+                {true, {ok, Module}} ->
+                    Module;
+                _ ->
+                    rabbit_misc:protocol_error(
+                      command_invalid,
+                      "invalid authentication mechanism '~s'", [T])
+            end
+    end.
+
+auth_mechanisms() ->
+    {ok, Configured} = application:get_env(auth_mechanisms),
+    [Name || {Name, _Module} <- rabbit_registry:lookup_all(auth_mechanism),
+             lists:member(Name, Configured)].
+
+auth_mechanisms_binary() ->
+    list_to_binary(
+            string:join(
+              [atom_to_list(A) || A <- auth_mechanisms()], " ")).
+
+auth_phase(Response,
+           State = #v1{auth_mechanism = AuthMechanism,
+                       auth_state = AuthState,
+                       connection = Connection =
+                           #connection{protocol = Protocol},
+                       sock = Sock}) ->
+    case AuthMechanism:handle_response(Response, AuthState) of
+        {refused, Reason} ->
+            rabbit_misc:protocol_error(
+              access_refused, "~s login refused: ~s",
+              [proplists:get_value(name, AuthMechanism:description()), Reason]);
+        {protocol_error, Msg, Args} ->
+            rabbit_misc:protocol_error(syntax_error, Msg, Args);
+        {challenge, Challenge, AuthState1} ->
+            Secure = #'connection.secure'{challenge = Challenge},
+            ok = send_on_channel0(Sock, Secure, Protocol),
+            State#v1{auth_state = AuthState1};
+        {ok, User} ->
+            Tune = #'connection.tune'{channel_max = 0,
+                                      frame_max = ?FRAME_MAX,
+                                      heartbeat = 0},
+            ok = send_on_channel0(Sock, Tune, Protocol),
+            State#v1{connection_state = tuning,
+                     connection = Connection#connection{user = User}}
+    end.
+
 %%--------------------------------------------------------------------------
 
 infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
@@ -864,6 +925,10 @@ i(protocol, #v1{connection = #connection{protocol = none}}) ->
     none;
 i(protocol, #v1{connection = #connection{protocol = Protocol}}) ->
     Protocol:version();
+i(auth_mechanism, #v1{auth_mechanism = none}) ->
+    none;
+i(auth_mechanism, #v1{auth_mechanism = Mechanism}) ->
+    proplists:get_value(name, Mechanism:description());
 i(user, #v1{connection = #connection{user = #user{username = Username}}}) ->
     Username;
 i(user, #v1{connection = #connection{user = none}}) ->
diff --git a/src/rabbit_exchange_type_registry.erl b/src/rabbit_registry.erl
index f15275b5..7a3fcb51 100644
--- a/src/rabbit_exchange_type_registry.erl
+++ b/src/rabbit_registry.erl
@@ -29,7 +29,7 @@
 %%   Contributor(s): ______________________________________.
 %%
 
--module(rabbit_exchange_type_registry).
+-module(rabbit_registry).
 
 -behaviour(gen_server).
 
@@ -38,7 +38,7 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          code_change/3]).
 
--export([register/2, binary_to_type/1, lookup_module/1]).
+-export([register/3, binary_to_type/1, lookup_module/2, lookup_all/1]).
 
 -define(SERVER, ?MODULE).
 -define(ETS_NAME, ?MODULE).
@@ -46,11 +46,12 @@
 -ifdef(use_specs).
 
 -spec(start_link/0 :: () -> rabbit_types:ok_pid_or_error()).
--spec(register/2 :: (binary(), atom()) -> 'ok').
+-spec(register/3 :: (atom(), binary(), atom()) -> 'ok').
 -spec(binary_to_type/1 ::
         (binary()) -> atom() | rabbit_types:error('not_found')).
--spec(lookup_module/1 ::
-        (atom()) -> rabbit_types:ok_or_error2(atom(), 'not_found')).
+-spec(lookup_module/2 ::
+        (atom(), atom()) -> rabbit_types:ok_or_error2(atom(), 'not_found')).
+-spec(lookup_all/1 :: (atom()) -> [{atom(), atom()}]).
 
 -endif.
 
@@ -61,8 +62,8 @@ start_link() ->
 
 %%---------------------------------------------------------------------------
 
-register(TypeName, ModuleName) ->
-    gen_server:call(?SERVER, {register, TypeName, ModuleName}).
+register(Class, TypeName, ModuleName) ->
+    gen_server:call(?SERVER, {register, Class, TypeName, ModuleName}).
 
 %% This is used with user-supplied arguments (e.g., on exchange
 %% declare), so we restrict it to existing atoms only.  This means it
@@ -74,47 +75,54 @@ binary_to_type(TypeBin) when is_binary(TypeBin) ->
         TypeAtom              -> TypeAtom
     end.
 
-lookup_module(T) when is_atom(T) ->
-    case ets:lookup(?ETS_NAME, T) of
+lookup_module(Class, T) when is_atom(T) ->
+    case ets:lookup(?ETS_NAME, {Class, T}) of
         [{_, Module}] ->
             {ok, Module};
         [] ->
             {error, not_found}
     end.
 
+lookup_all(Class) ->
+    [{K, V} || [K, V] <- ets:match(?ETS_NAME, {{Class, '$1'}, '$2'})].
+
 %%---------------------------------------------------------------------------
 
 internal_binary_to_type(TypeBin) when is_binary(TypeBin) ->
     list_to_atom(binary_to_list(TypeBin)).
 
-internal_register(TypeName, ModuleName)
-  when is_binary(TypeName), is_atom(ModuleName) ->
-    ok = sanity_check_module(ModuleName),
+internal_register(Class, TypeName, ModuleName)
+  when is_atom(Class), is_binary(TypeName), is_atom(ModuleName) ->
+    ok = sanity_check_module(class_module(Class), ModuleName),
     true = ets:insert(?ETS_NAME,
-                      {internal_binary_to_type(TypeName), ModuleName}),
+                      {{Class, internal_binary_to_type(TypeName)}, ModuleName}),
     ok.
 
-sanity_check_module(Module) ->
-    case catch lists:member(rabbit_exchange_type,
+sanity_check_module(ClassModule, Module) ->
+    case catch lists:member(ClassModule,
                             lists:flatten(
                               [Bs || {Attr, Bs} <-
                                          Module:module_info(attributes),
                                      Attr =:= behavior orelse
                                          Attr =:= behaviour])) of
         {'EXIT', {undef, _}}  -> {error, not_module};
-        false                 -> {error, not_exchange_type};
+        false                 -> {error, {not_type, ClassModule}};
         true                  -> ok
     end.
 
+class_module(exchange)       -> rabbit_exchange_type;
+class_module(auth_mechanism) -> rabbit_auth_mechanism.
+
 %%---------------------------------------------------------------------------
 
 init([]) ->
     ?ETS_NAME = ets:new(?ETS_NAME, [protected, set, named_table]),
     {ok, none}.
 
-handle_call({register, TypeName, ModuleName}, _From, State) ->
-    ok = internal_register(TypeName, ModuleName),
+handle_call({register, Class, TypeName, ModuleName}, _From, State) ->
+    ok = internal_register(Class, TypeName, ModuleName),
     {reply, ok, State};
+
 handle_call(Request, _From, State) ->
     {stop, {unhandled_call, Request}, State}.
 
diff --git a/src/rabbit_ssl.erl b/src/rabbit_ssl.erl
index 1d8ce23b..a4da23e2 100644
--- a/src/rabbit_ssl.erl
+++ b/src/rabbit_ssl.erl
@@ -36,6 +36,7 @@
 -include_lib("public_key/include/public_key.hrl").
 
 -export([peer_cert_issuer/1, peer_cert_subject/1, peer_cert_validity/1]).
+-export([peer_cert_subject_item/2]).
 
 %%--------------------------------------------------------------------------
 
@@ -45,9 +46,11 @@
 
 -type(certificate() :: binary()).
 
--spec(peer_cert_issuer/1   :: (certificate()) -> string()).
--spec(peer_cert_subject/1  :: (certificate()) -> string()).
--spec(peer_cert_validity/1 :: (certificate()) -> string()).
+-spec(peer_cert_issuer/1        :: (certificate()) -> string()).
+-spec(peer_cert_subject/1       :: (certificate()) -> string()).
+-spec(peer_cert_validity/1      :: (certificate()) -> string()).
+-spec(peer_cert_subject_item/2  ::
+        (certificate(), tuple()) -> string() | 'not_found').
 
 -endif.
 
@@ -71,6 +74,14 @@ peer_cert_subject(Cert) ->
                       format_rdn_sequence(Subject)
               end, Cert).
 
+%% Return a part of the certificate's subject.
+peer_cert_subject_item(Cert, Type) ->
+    cert_info(fun(#'OTPCertificate' {
+                     tbsCertificate = #'OTPTBSCertificate' {
+                       subject = Subject }}) ->
+                      find_by_type(Type, Subject)
+              end, Cert).
+
 %% Return a string describing the certificate's validity.
 peer_cert_validity(Cert) ->
     cert_info(fun(#'OTPCertificate' {
@@ -89,6 +100,14 @@ cert_info(F, Cert) ->
           DecCert       -> DecCert  %%R14B onwards
       end).
 
+find_by_type(Type, {rdnSequence, RDNs}) ->
+    case [V || #'AttributeTypeAndValue'{type = T, value = V}
+                     <- lists:flatten(RDNs),
+                 T == Type] of
+        [{printableString, S}] -> S;
+        []                     -> not_found
+    end.
+
 %%--------------------------------------------------------------------------
 %% Formatting functions
 %%--------------------------------------------------------------------------
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 8b58b822..adf968cb 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -35,8 +35,6 @@
 
 -export([all_tests/0, test_parsing/0]).
 
--import(lists).
-
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
 -include_lib("kernel/include/file.hrl").
@@ -1898,7 +1896,7 @@ test_variable_queue_ack_limiting(VQ0) ->
     VQ6 = check_variable_queue_status(
             rabbit_variable_queue:set_ram_duration_target(0, VQ5),
             [{len, Len div 2},
-             {target_ram_item_count, 0},
+             {target_ram_count, 0},
              {ram_msg_count, 0},
              {ram_ack_count, 0}]),
 
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 7c676164..bf7d2a5d 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -31,7 +31,7 @@
 
 -module(rabbit_variable_queue).
 
--export([init/5, init/3, terminate/1, delete_and_terminate/1,
+-export([init/3, terminate/1, delete_and_terminate/1,
          purge/1, publish/3, publish_delivered/4, fetch/2, ack/2,
          tx_publish/4, tx_ack/3, tx_rollback/2, tx_commit/4,
          requeue/3, len/1, is_empty/1, dropwhile/2,
@@ -42,7 +42,7 @@
 -export([start/1, stop/0]).
 
 %% exported for testing only
--export([start_msg_store/2, stop_msg_store/0]).
+-export([start_msg_store/2, stop_msg_store/0, init/5]).
 
 %%----------------------------------------------------------------------------
 %% Definitions:
@@ -158,7 +158,7 @@
 %% The conversion from alphas to betas is also chunked, but only to
 %% ensure no more than ?IO_BATCH_SIZE alphas are converted to betas at
 %% any one time. This further smooths the effects of changes to the
-%% target_ram_item_count and ensures the queue remains responsive
+%% target_ram_count and ensures the queue remains responsive
 %% even when there is a large amount of IO work to do. The
 %% idle_timeout callback is utilised to ensure that conversions are
 %% done as promptly as possible whilst ensuring the queue remains
@@ -256,7 +256,7 @@
           len,
           persistent_count,
 
-          target_ram_item_count,
+          target_ram_count,
           ram_msg_count,
           ram_msg_count_prev,
           ram_ack_count_prev,
@@ -323,7 +323,7 @@
                           timestamp   :: timestamp() }).
 
 -type(delta() :: #delta { start_seq_id :: non_neg_integer(),
-                          count        :: non_neg_integer (),
+                          count        :: non_neg_integer(),
                           end_seq_id   :: non_neg_integer() }).
 
 -type(sync() :: #sync { acks_persistent :: [[seq_id()]],
@@ -351,7 +351,7 @@
              persistent_count      :: non_neg_integer(),
 
              transient_threshold   :: non_neg_integer(),
-             target_ram_item_count :: non_neg_integer() | 'infinity',
+             target_ram_count      :: non_neg_integer() | 'infinity',
              ram_msg_count         :: non_neg_integer(),
              ram_msg_count_prev    :: non_neg_integer(),
              ram_index_count       :: non_neg_integer(),
@@ -734,26 +734,24 @@ len(#vqstate { len = Len }) -> Len.
 
 is_empty(State) -> 0 == len(State).
 
-set_ram_duration_target(DurationTarget,
-                        State = #vqstate {
-                          rates =
-                              #rates { avg_egress  = AvgEgressRate,
-                                       avg_ingress = AvgIngressRate },
-                          ack_rates =
-                              #rates { avg_egress  = AvgAckEgressRate,
-                                       avg_ingress = AvgAckIngressRate },
-                          target_ram_item_count = TargetRamItemCount }) ->
+set_ram_duration_target(
+  DurationTarget, State = #vqstate {
+                    rates     = #rates { avg_egress  = AvgEgressRate,
+                                         avg_ingress = AvgIngressRate },
+                    ack_rates = #rates { avg_egress  = AvgAckEgressRate,
+                                         avg_ingress = AvgAckIngressRate },
+                    target_ram_count = TargetRamCount }) ->
     Rate =
         AvgEgressRate + AvgIngressRate + AvgAckEgressRate + AvgAckIngressRate,
-    TargetRamItemCount1 =
+    TargetRamCount1 =
         case DurationTarget of
             infinity  -> infinity;
             _         -> trunc(DurationTarget * Rate) %% msgs = sec * msgs/sec
         end,
-    State1 = State #vqstate { target_ram_item_count = TargetRamItemCount1 },
-    a(case TargetRamItemCount1 == infinity orelse
-          (TargetRamItemCount =/= infinity andalso
-           TargetRamItemCount1 >= TargetRamItemCount) of
+    State1 = State #vqstate { target_ram_count = TargetRamCount1 },
+    a(case TargetRamCount1 == infinity orelse
+          (TargetRamCount =/= infinity andalso
+           TargetRamCount1 >= TargetRamCount) of
           true  -> State1;
           false -> reduce_memory_use(State1)
       end).
@@ -829,40 +827,39 @@ idle_timeout(State) -> a(reduce_memory_use(tx_commit_index(State))).
 handle_pre_hibernate(State = #vqstate { index_state = IndexState }) ->
     State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }.
 
-status(#vqstate { q1 = Q1, q2 = Q2, delta = Delta, q3 = Q3, q4 = Q4,
-                  len                   = Len,
-                  pending_ack           = PA,
-                  ram_ack_index         = RAI,
-                  on_sync               = #sync { funs = From },
-                  target_ram_item_count = TargetRamItemCount,
-                  ram_msg_count         = RamMsgCount,
-                  ram_index_count       = RamIndexCount,
-                  next_seq_id           = NextSeqId,
-                  persistent_count      = PersistentCount,
-                  rates                 = #rates {
-                    avg_egress  = AvgEgressRate,
-                    avg_ingress = AvgIngressRate },
-                  ack_rates             = #rates {
-                    avg_egress  = AvgAckEgressRate,
-                    avg_ingress = AvgAckIngressRate } }) ->
-    [ {q1                    , queue:len(Q1)},
-      {q2                    , bpqueue:len(Q2)},
-      {delta                 , Delta},
-      {q3                    , bpqueue:len(Q3)},
-      {q4                    , queue:len(Q4)},
-      {len                   , Len},
-      {pending_acks          , dict:size(PA)},
-      {outstanding_txns      , length(From)},
-      {target_ram_item_count , TargetRamItemCount},
-      {ram_msg_count         , RamMsgCount},
-      {ram_ack_count         , gb_trees:size(RAI)},
-      {ram_index_count       , RamIndexCount},
-      {next_seq_id           , NextSeqId},
-      {persistent_count      , PersistentCount},
-      {avg_ingress_rate      , AvgIngressRate},
-      {avg_egress_rate       , AvgEgressRate},
-      {avg_ack_ingress_rate  , AvgAckIngressRate},
-      {avg_ack_egress_rate   , AvgAckEgressRate} ].
+status(#vqstate {
+          q1 = Q1, q2 = Q2, delta = Delta, q3 = Q3, q4 = Q4,
+          len              = Len,
+          pending_ack      = PA,
+          ram_ack_index    = RAI,
+          on_sync          = #sync { funs = From },
+          target_ram_count = TargetRamCount,
+          ram_msg_count    = RamMsgCount,
+          ram_index_count  = RamIndexCount,
+          next_seq_id      = NextSeqId,
+          persistent_count = PersistentCount,
+          rates            = #rates { avg_egress  = AvgEgressRate,
+                                      avg_ingress = AvgIngressRate },
+          ack_rates        = #rates { avg_egress  = AvgAckEgressRate,
+                                      avg_ingress = AvgAckIngressRate } }) ->
+    [ {q1                  , queue:len(Q1)},
+      {q2                  , bpqueue:len(Q2)},
+      {delta               , Delta},
+      {q3                  , bpqueue:len(Q3)},
+      {q4                  , queue:len(Q4)},
+      {len                 , Len},
+      {pending_acks        , dict:size(PA)},
+      {outstanding_txns    , length(From)},
+      {target_ram_count    , TargetRamCount},
+      {ram_msg_count       , RamMsgCount},
+      {ram_ack_count       , gb_trees:size(RAI)},
+      {ram_index_count     , RamIndexCount},
+      {next_seq_id         , NextSeqId},
+      {persistent_count    , PersistentCount},
+      {avg_ingress_rate    , AvgIngressRate},
+      {avg_egress_rate     , AvgEgressRate},
+      {avg_ack_ingress_rate, AvgAckIngressRate},
+      {avg_ack_egress_rate , AvgAckEgressRate} ].
 
 %%----------------------------------------------------------------------------
 %% Minor helpers
@@ -1056,37 +1053,37 @@ init(IsDurable, IndexState, DeltaCount, Terms,
             end,
     Now = now(),
     State = #vqstate {
-      q1                    = queue:new(),
-      q2                    = bpqueue:new(),
-      delta                 = Delta,
-      q3                    = bpqueue:new(),
-      q4                    = queue:new(),
-      next_seq_id           = NextSeqId,
-      pending_ack           = dict:new(),
-      ram_ack_index         = gb_trees:empty(),
-      index_state           = IndexState1,
-      msg_store_clients     = {PersistentClient, TransientClient},
-      on_sync               = ?BLANK_SYNC,
-      durable               = IsDurable,
-      transient_threshold   = NextSeqId,
-
-      len                   = DeltaCount1,
-      persistent_count      = DeltaCount1,
-
-      target_ram_item_count = infinity,
-      ram_msg_count         = 0,
-      ram_msg_count_prev    = 0,
-      ram_ack_count_prev    = 0,
-      ram_index_count       = 0,
-      out_counter           = 0,
-      in_counter            = 0,
-      msgs_on_disk         = gb_sets:new(),
-      msg_indices_on_disk  = gb_sets:new(),
-      unconfirmed          = gb_sets:new(),
-      ack_out_counter       = 0,
-      ack_in_counter        = 0,
-      rates                 = blank_rate(Now, DeltaCount1),
-      ack_rates             = blank_rate(Now, 0) },
+      q1                  = queue:new(),
+      q2                  = bpqueue:new(),
+      delta               = Delta,
+      q3                  = bpqueue:new(),
+      q4                  = queue:new(),
+      next_seq_id         = NextSeqId,
+      pending_ack         = dict:new(),
+      ram_ack_index       = gb_trees:empty(),
+      index_state         = IndexState1,
+      msg_store_clients   = {PersistentClient, TransientClient},
+      on_sync             = ?BLANK_SYNC,
+      durable             = IsDurable,
+      transient_threshold = NextSeqId,
+
+      len                 = DeltaCount1,
+      persistent_count    = DeltaCount1,
+
+      target_ram_count    = infinity,
+      ram_msg_count       = 0,
+      ram_msg_count_prev  = 0,
+      ram_ack_count_prev  = 0,
+      ram_index_count     = 0,
+      out_counter         = 0,
+      in_counter          = 0,
+      rates               = blank_rate(Now, DeltaCount1),
+      msgs_on_disk        = gb_sets:new(),
+      msg_indices_on_disk = gb_sets:new(),
+      unconfirmed         = gb_sets:new(),
+      ack_out_counter     = 0,
+      ack_in_counter      = 0,
+      ack_rates           = blank_rate(Now, 0) },
     a(maybe_deltas_to_betas(State)).
 
 blank_rate(Timestamp, IngressLength) ->
@@ -1443,7 +1440,7 @@ msg_indices_written_to_disk(QPid, GuidSet) ->
 %% though the conversion function for that is called as necessary. The
 %% reason is twofold. Firstly, this is safe because the conversion is
 %% only ever necessary just after a transition to a
-%% target_ram_item_count of zero or after an incremental alpha->beta
+%% target_ram_count of zero or after an incremental alpha->beta
 %% conversion. In the former case the conversion is performed straight
 %% away (i.e. any betas present at the time are converted to deltas),
 %% and in the latter case the need for a conversion is flagged up
@@ -1454,51 +1451,41 @@ msg_indices_written_to_disk(QPid, GuidSet) ->
 %% perpetually reporting the need for a conversion when no such
 %% conversion is needed. That in turn could cause an infinite loop.
 reduce_memory_use(_AlphaBetaFun, _BetaGammaFun, _BetaDeltaFun, _AckFun,
-                  State = #vqstate {target_ram_item_count = infinity}) ->
+                  State = #vqstate {target_ram_count = infinity}) ->
     {false, State};
 reduce_memory_use(AlphaBetaFun, BetaGammaFun, BetaDeltaFun, AckFun,
                   State = #vqstate {
-                    ram_ack_index         = RamAckIndex,
-                    ram_msg_count         = RamMsgCount,
-                    target_ram_item_count = TargetRamItemCount,
-                    rates                 = #rates {
-                      avg_ingress = AvgIngress,
-                      avg_egress  = AvgEgress },
-                    ack_rates             = #rates {
-                      avg_ingress = AvgAckIngress,
-                      avg_egress  = AvgAckEgress } }) ->
+                    ram_ack_index    = RamAckIndex,
+                    ram_msg_count    = RamMsgCount,
+                    target_ram_count = TargetRamCount,
+                    rates            = #rates { avg_ingress = AvgIngress,
+                                                avg_egress  = AvgEgress },
+                    ack_rates        = #rates { avg_ingress = AvgAckIngress,
+                                                avg_egress  = AvgAckEgress }
+                   }) ->
 
     {Reduce, State1} =
         case chunk_size(RamMsgCount + gb_trees:size(RamAckIndex),
-                        TargetRamItemCount) of
-            0 ->
-                {false, State};
-            S1 ->
-                ReduceFuns =
-                    case (AvgAckIngress - AvgAckEgress) >
-                        (AvgIngress - AvgEgress) of
-                        true ->
-                            %% ACKs are growing faster than the queue,
-                            %% push messages from there first.
-                            [AckFun, AlphaBetaFun];
-                        false ->
-                            %% The queue is growing faster than the
-                            %% acks, push queue messages first.
-                            [AlphaBetaFun, AckFun]
-                    end,
-                {_, State2} =
-                    %% Both reduce functions get a chance to reduce
-                    %% memory. The second may very well get a quota of
-                    %% 0 if the first function managed to push out the
-                    %% maximum number of messages.
-                    lists:foldl(
-                      fun (ReduceFun, {QuotaN, StateN}) ->
-                              ReduceFun(QuotaN, StateN)
-                      end, {S1, State}, ReduceFuns),
-                {true, State2}
+                        TargetRamCount) of
+            0  -> {false, State};
+            %% Reduce memory of pending acks and alphas. The order is
+            %% determined based on which is growing faster. Whichever
+            %% comes second may very well get a quota of 0 if the
+            %% first manages to push out the max number of messages.
+            S1 -> {_, State2} =
+                      lists:foldl(fun (ReduceFun, {QuotaN, StateN}) ->
+                                          ReduceFun(QuotaN, StateN)
+                                  end,
+                                  {S1, State},
+                                  case (AvgAckIngress - AvgAckEgress) >
+                                      (AvgIngress - AvgEgress) of
+                                      true  -> [AckFun, AlphaBetaFun];
+                                      false -> [AlphaBetaFun, AckFun]
+                                  end),
+                  {true, State2}
         end,
 
-    case State1 #vqstate.target_ram_item_count of
+    case State1 #vqstate.target_ram_count of
         0 -> {Reduce, BetaDeltaFun(State1)};
         _ -> case chunk_size(State1 #vqstate.ram_index_count,
                              permitted_ram_index_count(State1)) of
@@ -1694,11 +1681,11 @@ maybe_push_q4_to_betas(Quota, State = #vqstate { q4 = Q4 }) ->
 
 maybe_push_alphas_to_betas(_Generator, _Consumer, Quota, _Q,
                            State = #vqstate {
-                             ram_msg_count         = RamMsgCount,
-                             target_ram_item_count = TargetRamItemCount })
+                             ram_msg_count    = RamMsgCount,
+                             target_ram_count = TargetRamCount })
   when Quota =:= 0 orelse
-       TargetRamItemCount =:= infinity orelse
-       TargetRamItemCount >= RamMsgCount ->
+       TargetRamCount =:= infinity orelse
+       TargetRamCount >= RamMsgCount ->
     {Quota, State};
 maybe_push_alphas_to_betas(Generator, Consumer, Quota, Q, State) ->
     case Generator(Q) of
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index 0159609d..068ac186 100644
--- a/src/rabbit_writer.erl
+++ b/src/rabbit_writer.erl
@@ -39,8 +39,6 @@
          send_command_and_notify/4, send_command_and_notify/5]).
 -export([internal_send_command/4, internal_send_command/6]).
 
--import(gen_tcp).
-
 -record(wstate, {sock, channel, frame_max, protocol}).
 
 -define(HIBERNATE_AFTER, 5000).