Merged bug23331 into default

41 files changed, 1173 insertions, 718 deletions

diff --git a/Makefile b/Makefile
index 38ec8196de..ee0c083863 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,3 @@
-
 TMPDIR ?= /tmp
 
 RABBITMQ_NODENAME ?= rabbit
@@ -92,12 +91,13 @@ endif
 all: $(TARGETS)
 
 $(DEPS_FILE): $(SOURCES) $(INCLUDES)
+	rm -f $@
 	escript generate_deps $(INCLUDE_DIR) $(SOURCE_DIR) \$$\(EBIN_DIR\) $@
 
 $(EBIN_DIR)/rabbit.app: $(EBIN_DIR)/rabbit_app.in $(BEAM_TARGETS) generate_app
 	escript generate_app $(EBIN_DIR) $@ < $<
 
-$(EBIN_DIR)/%.beam: $(SOURCE_DIR)/%.erl $(DEPS_FILE)
+$(EBIN_DIR)/%.beam: $(SOURCE_DIR)/%.erl | $(DEPS_FILE)
 	erlc $(ERLC_OPTS) -pa $(EBIN_DIR) $<
 
 $(INCLUDE_DIR)/rabbit_framing.hrl: codegen.py $(AMQP_CODEGEN_DIR)/amqp_codegen.py $(AMQP_SPEC_JSON_FILES_0_9_1) $(AMQP_SPEC_JSON_FILES_0_8)
diff --git a/codegen.py b/codegen.py
index 142297533f..4fdbec554f 100644
--- a/codegen.py
+++ b/codegen.py
@@ -75,6 +75,8 @@ def erlangize(s):
 
 AmqpMethod.erlangName = lambda m: "'" + erlangize(m.klass.name) + '.' + erlangize(m.name) + "'"
 
+AmqpClass.erlangName = lambda c: "'" + erlangize(c.name) + "'"
+
 def erlangConstantName(s):
     return '_'.join(re.split('[- ]', s.upper()))
 
@@ -167,6 +169,9 @@ def genErl(spec):
     def genLookupMethodName(m):
         print "lookup_method_name({%d, %d}) -> %s;" % (m.klass.index, m.index, m.erlangName())
 
+    def genLookupClassName(c):
+        print "lookup_class_name(%d) -> %s;" % (c.index, c.erlangName())
+
     def genMethodId(m):
         print "method_id(%s) -> {%d, %d};" % (m.erlangName(), m.klass.index, m.index)
 
@@ -325,6 +330,8 @@ def genErl(spec):
 
 -export([version/0]).
 -export([lookup_method_name/1]).
+-export([lookup_class_name/1]).
+
 -export([method_id/1]).
 -export([method_has_content/1]).
 -export([is_method_synchronous/1]).
@@ -427,6 +434,9 @@ bitvalue(undefined) -> 0.
     for m in methods: genLookupMethodName(m)
     print "lookup_method_name({_ClassId, _MethodId} = Id) -> exit({unknown_method_id, Id})."
 
+    for c in spec.allClasses(): genLookupClassName(c)
+    print "lookup_class_name(ClassId) -> exit({unknown_class_id, ClassId})."
+
     for m in methods: genMethodId(m)
     print "method_id(Name) -> exit({unknown_method_name, Name})."
 
diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml
index 5179eb253c..3b7244c701 100644
--- a/docs/rabbitmqctl.1.xml
+++ b/docs/rabbitmqctl.1.xml
@@ -894,16 +894,31 @@
             </para>
             <variablelist>
               <varlistentry>
-                <term>exchange_name</term>
-                <listitem><para>The name of the exchange to which the
-                binding is attached. with non-ASCII characters
-                escaped as in C.</para></listitem>
+                <term>source_name</term>
+                <listitem><para>The name of the source of messages to
+                which the binding is attached. With non-ASCII
+                characters escaped as in C.</para></listitem>
               </varlistentry>
               <varlistentry>
-                <term>queue_name</term>
-                <listitem><para>The name of the queue to which the
-                binding is attached. with non-ASCII characters
-                escaped as in C.</para></listitem>
+                <term>source_kind</term>
+                <listitem><para>The kind of the source of messages to
+                which the binding is attached. Currently always
+                queue. With non-ASCII characters escaped as in
+                C.</para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>destination_name</term>
+                <listitem><para>The name of the destination of
+                messages to which the binding is attached. With
+                non-ASCII characters escaped as in
+                C.</para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>destination_kind</term>
+                <listitem><para>The kind of the destination of
+                messages to which the binding is attached. With
+                non-ASCII characters escaped as in
+                C.</para></listitem>
               </varlistentry>
               <varlistentry>
                 <term>routing_key</term>
@@ -967,6 +982,21 @@
                 <listitem><para>Peer port.</para></listitem>
               </varlistentry>
               <varlistentry>
+                <term>peer_cert_subject</term>
+                <listitem><para>The subject of the peer's SSL
+                certificate, in RFC4514 form.</para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>peer_cert_issuer</term>
+                <listitem><para>The issuer of the peer's SSL
+                certificate, in RFC4514 form.</para></listitem>
+              </varlistentry>
+              <varlistentry>
+                <term>peer_cert_validity</term>
+                <listitem><para>The period for which the peer's SSL
+                certificate is valid.</para></listitem>
+              </varlistentry>
+              <varlistentry>
                 <term>state</term>
                 <listitem><para>Connection state (one of [<command>starting</command>, <command>tuning</command>,
                   <command>opening</command>, <command>running</command>, <command>closing</command>, <command>closed</command>]).</para></listitem>
@@ -1101,6 +1131,14 @@
                 <term>prefetch_count</term>
                 <listitem><para>QoS prefetch count limit in force, 0 if unlimited.</para></listitem>
               </varlistentry>
+              <varlistentry>
+                <term>client_flow_blocked</term>
+                <listitem><para>True if the client issued a
+                    <command>channel.flow{active=false}</command>
+                    command, blocking the server from delivering
+                    messages to the channel's consumers.
+                    </para></listitem>
+              </varlistentry>
             </variablelist>
             <para>
               If no <command>channelinfoitem</command>s are specified then pid,
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 24aa8d987c..af6e257ade 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -60,8 +60,8 @@
 -record(route, {binding, value = const}).
 -record(reverse_route, {reverse_binding, value = const}).
 
--record(binding, {exchange_name, key, queue_name, args = []}).
--record(reverse_binding, {queue_name, key, exchange_name, args = []}).
+-record(binding, {source, key, destination, args = []}).
+-record(reverse_binding, {destination, key, source, args = []}).
 
 -record(listener, {node, protocol, host, port}).
 
@@ -70,7 +70,7 @@
 
 -record(ssl_socket, {tcp, ssl}).
 -record(delivery, {mandatory, immediate, txn, sender, message}).
--record(amqp_error, {name, explanation, method = none}).
+-record(amqp_error, {name, explanation = "", method = none}).
 
 -record(event, {type, props, timestamp}).
 
diff --git a/include/rabbit_backing_queue_spec.hrl b/include/rabbit_backing_queue_spec.hrl
index 005994f09f..38c6f9398f 100644
--- a/include/rabbit_backing_queue_spec.hrl
+++ b/include/rabbit_backing_queue_spec.hrl
@@ -30,8 +30,9 @@
 %%
 
 -type(fetch_result() ::
-                 %% Message,  IsDelivered,  AckTag,  Remaining_Len
-        ('empty'|{rabbit_types:basic_message(), boolean(), ack(), non_neg_integer()})).
+        ('empty' |
+         %% Message,  IsDelivered,  AckTag,  RemainingLen
+         {rabbit_types:basic_message(), boolean(), ack(), non_neg_integer()})).
 -type(is_durable() :: boolean()).
 -type(attempt_recovery() :: boolean()).
 -type(purged_msg_count() :: non_neg_integer()).
@@ -39,19 +40,23 @@
 
 -spec(start/1 :: ([rabbit_amqqueue:name()]) -> 'ok').
 -spec(stop/0 :: () -> 'ok').
--spec(init/3 :: (rabbit_amqqueue:name(), is_durable(), attempt_recovery()) -> state()).
+-spec(init/3 :: (rabbit_amqqueue:name(), is_durable(), attempt_recovery()) ->
+                     state()).
 -spec(terminate/1 :: (state()) -> state()).
 -spec(delete_and_terminate/1 :: (state()) -> state()).
 -spec(purge/1 :: (state()) -> {purged_msg_count(), state()}).
 -spec(publish/2 :: (rabbit_types:basic_message(), state()) -> state()).
 -spec(publish_delivered/3 ::
-        (ack_required(), rabbit_types:basic_message(), state()) -> {ack(), state()}).
+        (ack_required(), rabbit_types:basic_message(), state()) ->
+                                  {ack(), state()}).
 -spec(fetch/2 :: (ack_required(), state()) -> {fetch_result(), state()}).
 -spec(ack/2 :: ([ack()], state()) -> state()).
--spec(tx_publish/3 :: (rabbit_types:txn(), rabbit_types:basic_message(), state()) -> state()).
+-spec(tx_publish/3 :: (rabbit_types:txn(), rabbit_types:basic_message(),
+                       state()) -> state()).
 -spec(tx_ack/3 :: (rabbit_types:txn(), [ack()], state()) -> state()).
 -spec(tx_rollback/2 :: (rabbit_types:txn(), state()) -> {[ack()], state()}).
--spec(tx_commit/3 :: (rabbit_types:txn(), fun (() -> any()), state()) -> {[ack()], state()}).
+-spec(tx_commit/3 :: (rabbit_types:txn(), fun (() -> any()), state()) ->
+                          {[ack()], state()}).
 -spec(requeue/2 :: ([ack()], state()) -> state()).
 -spec(len/1 :: (state()) -> non_neg_integer()).
 -spec(is_empty/1 :: (state()) -> boolean()).
diff --git a/include/rabbit_exchange_type_spec.hrl b/include/rabbit_exchange_type_spec.hrl
index cecd666b4d..ae326a872d 100644
--- a/include/rabbit_exchange_type_spec.hrl
+++ b/include/rabbit_exchange_type_spec.hrl
@@ -31,8 +31,8 @@
 -ifdef(use_specs).
 
 -spec(description/0 :: () -> [{atom(), any()}]).
--spec(publish/2 :: (rabbit_types:exchange(), rabbit_types:delivery())
-                   -> {rabbit_router:routing_result(), [pid()]}).
+-spec(route/2 :: (rabbit_types:exchange(), rabbit_types:delivery())
+                 -> rabbit_router:match_result()).
 -spec(validate/1 :: (rabbit_types:exchange()) -> 'ok').
 -spec(create/1 :: (rabbit_types:exchange()) -> 'ok').
 -spec(recover/2 :: (rabbit_types:exchange(),
diff --git a/include/rabbit_msg_store_index.hrl b/include/rabbit_msg_store_index.hrl
index fba0b7cd4e..d4115363cb 100644
--- a/include/rabbit_msg_store_index.hrl
+++ b/include/rabbit_msg_store_index.hrl
@@ -51,6 +51,7 @@
                                                [{fieldpos(), fieldvalue()}]),
                           index_state()) -> 'ok').
 -spec(delete/2 :: (rabbit_guid:guid(), index_state()) -> 'ok').
+-spec(delete_object/2 :: (keyvalue(), index_state()) -> 'ok').
 -spec(delete_by_file/2 :: (fieldvalue(), index_state()) -> 'ok').
 -spec(terminate/1 :: (index_state()) -> any()).
 
diff --git a/packaging/RPMS/Fedora/rabbitmq-server.spec b/packaging/RPMS/Fedora/rabbitmq-server.spec
index eb0a2a5101..209a90ee31 100644
--- a/packaging/RPMS/Fedora/rabbitmq-server.spec
+++ b/packaging/RPMS/Fedora/rabbitmq-server.spec
@@ -127,6 +127,9 @@ done
 rm -rf %{buildroot}
 
 %changelog
+* Tue Oct 19 2010 vlad@rabbitmq.com 2.1.1-1
+- New Upstream Release
+
 * Tue Sep 14 2010 marek@rabbitmq.com 2.1.0-1
 - New Upstream Release
 
diff --git a/packaging/debs/Debian/debian/changelog b/packaging/debs/Debian/debian/changelog
index 9927cfbcfd..e81fda241f 100644
--- a/packaging/debs/Debian/debian/changelog
+++ b/packaging/debs/Debian/debian/changelog
@@ -1,3 +1,9 @@
+rabbitmq-server (2.1.1-1) lucid; urgency=low
+
+  * New Upstream Release
+
+ -- Vlad Alexandru Ionescu <vlad@rabbitmq.com>  Tue, 19 Oct 2010 17:20:10 +0100
+
 rabbitmq-server (2.1.0-1) lucid; urgency=low
 
   * New Upstream Release
diff --git a/src/file_handle_cache.erl b/src/file_handle_cache.erl
index d2830a25ca..6a948d49b6 100644
--- a/src/file_handle_cache.erl
+++ b/src/file_handle_cache.erl
@@ -1171,7 +1171,7 @@ ulimit() ->
             ?FILE_HANDLES_LIMIT_WINDOWS;
         {unix, _OsName} ->
             %% Under Linux, Solaris and FreeBSD, ulimit is a shell
-            %% builtin, not a command. In OS X, it's a command.
+            %% builtin, not a command. In OS X and AIX it's a command.
             %% Fortunately, os:cmd invokes the cmd in a shell env, so
             %% we're safe in all cases.
             case os:cmd("ulimit -n") of
diff --git a/src/gen_server2.erl b/src/gen_server2.erl
index b0379b95d1..230d1f2aa1 100644
--- a/src/gen_server2.erl
+++ b/src/gen_server2.erl
@@ -1128,7 +1128,8 @@ function_exported_or_default(Mod, Fun, Arity, Default) ->
 %%-----------------------------------------------------------------
 format_status(Opt, StatusData) ->
     [PDict, SysState, Parent, Debug,
-     [Name, State, Mod, _Time, _TimeoutState, Queue]] = StatusData,
+     #gs2_state{name = Name, state = State, mod = Mod, queue = Queue}] =
+        StatusData,
     NameTag = if is_pid(Name) ->
                       pid_to_list(Name);
                  is_atom(Name) ->
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 3e677c3809..2389ec86df 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -31,7 +31,7 @@
 
 -module(rabbit_amqqueue).
 
--export([start/0, stop/0, declare/5, delete/3, purge/1]).
+-export([start/0, stop/0, declare/5, delete_immediately/1, delete/3, purge/1]).
 -export([internal_declare/2, internal_delete/1,
          maybe_run_queue_via_backing_queue/2,
          update_ram_duration/1, set_ram_duration_target/2,
@@ -115,6 +115,7 @@
         (rabbit_types:amqqueue())
         -> {'ok', non_neg_integer(), non_neg_integer()}).
 -spec(emit_stats/1 :: (rabbit_types:amqqueue()) -> 'ok').
+-spec(delete_immediately/1 :: (rabbit_types:amqqueue()) -> 'ok').
 -spec(delete/3 ::
       (rabbit_types:amqqueue(), 'false', 'false')
         -> qlen();
@@ -251,10 +252,10 @@ start_queue_process(Q) ->
 add_default_binding(#amqqueue{name = QueueName}) ->
     ExchangeName = rabbit_misc:r(QueueName, exchange, <<>>),
     RoutingKey = QueueName#resource.name,
-    rabbit_binding:add(#binding{exchange_name = ExchangeName,
-                                queue_name    = QueueName,
-                                key           = RoutingKey,
-                                args          = []}).
+    rabbit_binding:add(#binding{source      = ExchangeName,
+                                destination = QueueName,
+                                key         = RoutingKey,
+                                args        = []}).
 
 lookup(Name) ->
     rabbit_misc:dirty_read({rabbit_queue, Name}).
@@ -359,6 +360,9 @@ stat(#amqqueue{pid = QPid}) -> delegate_call(QPid, stat, infinity).
 emit_stats(#amqqueue{pid = QPid}) ->
     delegate_cast(QPid, emit_stats).
 
+delete_immediately(#amqqueue{ pid = QPid }) ->
+    gen_server2:cast(QPid, delete_immediately).
+
 delete(#amqqueue{ pid = QPid }, IfUnused, IfEmpty) ->
     delegate_call(QPid, {delete, IfUnused, IfEmpty}, infinity).
 
@@ -431,24 +435,20 @@ flush_all(QPids, ChPid) ->
 internal_delete1(QueueName) ->
     ok = mnesia:delete({rabbit_queue, QueueName}),
     ok = mnesia:delete({rabbit_durable_queue, QueueName}),
-    %% we want to execute some things, as
-    %% decided by rabbit_exchange, after the
-    %% transaction.
-    rabbit_binding:remove_for_queue(QueueName).
+    %% we want to execute some things, as decided by rabbit_exchange,
+    %% after the transaction.
+    rabbit_binding:remove_for_destination(QueueName).
 
 internal_delete(QueueName) ->
-    case
-        rabbit_misc:execute_mnesia_transaction(
-          fun () ->
-                  case mnesia:wread({rabbit_queue, QueueName}) of
-                      []  -> {error, not_found};
-                      [_] -> internal_delete1(QueueName)
-                  end
-          end) of
-        Err = {error, _} -> Err;
-        PostHook ->
-            PostHook(),
-            ok
+    case rabbit_misc:execute_mnesia_transaction(
+           fun () ->
+                   case mnesia:wread({rabbit_queue, QueueName}) of
+                       []  -> {error, not_found};
+                       [_] -> internal_delete1(QueueName)
+                   end
+           end) of
+        {error, _} = Err -> Err;
+        Deletions        -> ok = rabbit_binding:process_deletions(Deletions)
     end.
 
 maybe_run_queue_via_backing_queue(QPid, Fun) ->
@@ -467,20 +467,20 @@ maybe_expire(QPid) ->
     gen_server2:cast(QPid, maybe_expire).
 
 on_node_down(Node) ->
-    [Hook() ||
-        Hook <- rabbit_misc:execute_mnesia_transaction(
-                  fun () ->
-                          qlc:e(qlc:q([delete_queue(QueueName) ||
-                                          #amqqueue{name = QueueName, pid = Pid}
-                                              <- mnesia:table(rabbit_queue),
-                                          node(Pid) == Node]))
-                  end)],
-    ok.
+    rabbit_binding:process_deletions(
+      lists:foldl(
+        fun rabbit_binding:combine_deletions/2,
+        rabbit_binding:new_deletions(),
+        rabbit_misc:execute_mnesia_transaction(
+          fun () -> qlc:e(qlc:q([delete_queue(QueueName) ||
+                                    #amqqueue{name = QueueName, pid = Pid}
+                                        <- mnesia:table(rabbit_queue),
+                                    node(Pid) == Node]))
+          end))).
 
 delete_queue(QueueName) ->
-    Post = rabbit_binding:remove_transient_for_queue(QueueName),
     ok = mnesia:delete({rabbit_queue, QueueName}),
-    Post.
+    rabbit_binding:remove_transient_for_destination(QueueName).
 
 pseudo_queue(QueueName, Pid) ->
     #amqqueue{name = QueueName,
@@ -506,4 +506,3 @@ delegate_call(Pid, Msg, Timeout) ->
 
 delegate_cast(Pid, Msg) ->
     delegate:invoke(Pid, fun (P) -> gen_server2:cast(P, Msg) end).
-
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index d15a6eb3a9..19db731a2f 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -43,7 +43,7 @@
 
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
          handle_info/2, handle_pre_hibernate/1, prioritise_call/3,
-         prioritise_cast/2]).
+         prioritise_cast/2, prioritise_info/2]).
 
 -import(queue).
 -import(erlang).
@@ -251,8 +251,9 @@ stop_expiry_timer(State = #q{expiry_timer_ref = TRef}) ->
     {ok, cancel} = timer:cancel(TRef),
     State#q{expiry_timer_ref = undefined}.
 
-%% We only wish to expire where there are no consumers *and* when
-%% basic.get hasn't been called for the configured period.
+%% We wish to expire only when there are no consumers *and* the expiry
+%% hasn't been refreshed (by queue.declare or basic.get) for the
+%% configured period.
 ensure_expiry_timer(State = #q{expires = undefined}) ->
     State;
 ensure_expiry_timer(State = #q{expires = Expires}) ->
@@ -600,6 +601,7 @@ prioritise_call(Msg, _From, _State) ->
 prioritise_cast(Msg, _State) ->
     case Msg of
         update_ram_duration                  -> 8;
+        delete_immediately                   -> 8;
         {set_ram_duration_target, _Duration} -> 8;
         {set_maximum_since_use, _Age}        -> 8;
         maybe_expire                         -> 8;
@@ -611,6 +613,10 @@ prioritise_cast(Msg, _State) ->
         _                                    -> 0
     end.
 
+prioritise_info({'DOWN', _MonitorRef, process, DownPid, _Reason},
+                #q{q = #amqqueue{exclusive_owner = DownPid}}) -> 8;
+prioritise_info(_Msg, _State)                                 -> 0.
+
 handle_call({init, Recover}, From,
             State = #q{q = #amqqueue{exclusive_owner = none}}) ->
     declare(Recover, From, State);
@@ -778,7 +784,8 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
 handle_call(stat, _From, State = #q{backing_queue = BQ,
                                     backing_queue_state = BQS,
                                     active_consumers = ActiveConsumers}) ->
-    reply({ok, BQ:len(BQS), queue:len(ActiveConsumers)}, State);
+    reply({ok, BQ:len(BQS), queue:len(ActiveConsumers)},
+          ensure_expiry_timer(State));
 
 handle_call({delete, IfUnused, IfEmpty}, _From,
             State = #q{backing_queue_state = BQS, backing_queue = BQ}) ->
@@ -851,6 +858,9 @@ handle_cast({reject, AckTags, Requeue, ChPid},
 handle_cast({rollback, Txn, ChPid}, State) ->
     noreply(rollback_transaction(Txn, ChPid, State));
 
+handle_cast(delete_immediately, State) ->
+    {stop, normal, State};
+
 handle_cast({unblock, ChPid}, State) ->
     noreply(
       possibly_unblock(State, ChPid,
diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
index d62fc07cb0..3841298244 100644
--- a/src/rabbit_basic.erl
+++ b/src/rabbit_basic.erl
@@ -55,28 +55,24 @@
          rabbit_types:message()) -> rabbit_types:delivery()).
 -spec(message/4 ::
         (rabbit_exchange:name(), rabbit_router:routing_key(),
-         properties_input(), binary())
-        -> (rabbit_types:message() | rabbit_types:error(any()))).
+         properties_input(), binary()) ->
+                        (rabbit_types:message() | rabbit_types:error(any()))).
 -spec(properties/1 ::
         (properties_input()) -> rabbit_framing:amqp_property_record()).
 -spec(publish/4 ::
         (rabbit_exchange:name(), rabbit_router:routing_key(),
-         properties_input(), binary())
-        -> publish_result()).
+         properties_input(), binary()) -> publish_result()).
 -spec(publish/7 ::
         (rabbit_exchange:name(), rabbit_router:routing_key(),
          boolean(), boolean(), rabbit_types:maybe(rabbit_types:txn()),
-         properties_input(), binary())
-        -> publish_result()).
--spec(build_content/2 ::
-        (rabbit_framing:amqp_property_record(), binary())
-        -> rabbit_types:content()).
--spec(from_content/1 ::
-        (rabbit_types:content())
-        -> {rabbit_framing:amqp_property_record(), binary()}).
--spec(is_message_persistent/1 ::
-        (rabbit_types:decoded_content())
-        -> (boolean() | {'invalid', non_neg_integer()})).
+         properties_input(), binary()) -> publish_result()).
+-spec(build_content/2 :: (rabbit_framing:amqp_property_record(), binary()) ->
+                              rabbit_types:content()).
+-spec(from_content/1 :: (rabbit_types:content()) ->
+                             {rabbit_framing:amqp_property_record(), binary()}).
+-spec(is_message_persistent/1 :: (rabbit_types:decoded_content()) ->
+                                      (boolean() |
+                                       {'invalid', non_neg_integer()})).
 
 -endif.
 
diff --git a/src/rabbit_binary_generator.erl b/src/rabbit_binary_generator.erl
index 056ab1b574..722573c769 100644
--- a/src/rabbit_binary_generator.erl
+++ b/src/rabbit_binary_generator.erl
@@ -47,6 +47,7 @@
 -export([generate_table/1, encode_properties/2]).
 -export([check_empty_content_body_frame_size/0]).
 -export([ensure_content_encoded/2, clear_encoded_content/1]).
+-export([map_exception/3]).
 
 -import(lists).
 
@@ -74,6 +75,9 @@
                                        rabbit_types:encoded_content()).
 -spec(clear_encoded_content/1 ::
         (rabbit_types:content()) -> rabbit_types:unencoded_content()).
+-spec(map_exception/3 :: (non_neg_integer(), rabbit_types:amqp_error(),
+                          rabbit_types:protocol()) ->
+          {boolean(), non_neg_integer(), rabbit_framing:amqp_method()}).
 
 -endif.
 
@@ -306,3 +310,46 @@ clear_encoded_content(Content = #content{properties = none}) ->
     Content;
 clear_encoded_content(Content = #content{}) ->
     Content#content{properties_bin = none, protocol = none}.
+
+%% NB: this function is also used by the Erlang client
+map_exception(Channel, Reason, Protocol) ->
+    {SuggestedClose, ReplyCode, ReplyText, FailedMethod} =
+        lookup_amqp_exception(Reason, Protocol),
+    ShouldClose = SuggestedClose orelse (Channel == 0),
+    {ClassId, MethodId} = case FailedMethod of
+                              {_, _} -> FailedMethod;
+                              none   -> {0, 0};
+                              _      -> Protocol:method_id(FailedMethod)
+                          end,
+    {CloseChannel, CloseMethod} =
+        case ShouldClose of
+            true  -> {0, #'connection.close'{reply_code = ReplyCode,
+                                             reply_text = ReplyText,
+                                             class_id = ClassId,
+                                             method_id = MethodId}};
+            false -> {Channel, #'channel.close'{reply_code = ReplyCode,
+                                                reply_text = ReplyText,
+                                                class_id = ClassId,
+                                                method_id = MethodId}}
+        end,
+    {ShouldClose, CloseChannel, CloseMethod}.
+
+lookup_amqp_exception(#amqp_error{name        = Name,
+                                  explanation = Expl,
+                                  method      = Method},
+                      Protocol) ->
+    {ShouldClose, Code, Text} = Protocol:lookup_amqp_exception(Name),
+    ExplBin = amqp_exception_explanation(Text, Expl),
+    {ShouldClose, Code, ExplBin, Method};
+lookup_amqp_exception(Other, Protocol) ->
+    rabbit_log:warning("Non-AMQP exit reason '~p'~n", [Other]),
+    {ShouldClose, Code, Text} =
+        Protocol:lookup_amqp_exception(internal_error, Protocol),
+    {ShouldClose, Code, Text, none}.
+
+amqp_exception_explanation(Text, Expl) ->
+    ExplBin = list_to_binary(Expl),
+    CompleteTextBin = <<Text/binary, " - ", ExplBin/binary>>,
+    if size(CompleteTextBin) > 255 -> <<CompleteTextBin:252/binary, "...">>;
+       true                        -> CompleteTextBin
+    end.
diff --git a/src/rabbit_binding.erl b/src/rabbit_binding.erl
index 19150fa9f9..1af213c474 100644
--- a/src/rabbit_binding.erl
+++ b/src/rabbit_binding.erl
@@ -33,29 +33,35 @@
 -include("rabbit.hrl").
 
 -export([recover/0, exists/1, add/1, remove/1, add/2, remove/2, list/1]).
--export([list_for_exchange/1, list_for_queue/1, list_for_exchange_and_queue/2]).
+-export([list_for_source/1, list_for_destination/1,
+         list_for_source_and_destination/2]).
+-export([new_deletions/0, combine_deletions/2, add_deletion/3,
+         process_deletions/1]).
 -export([info_keys/0, info/1, info/2, info_all/1, info_all/2]).
 %% these must all be run inside a mnesia tx
--export([has_for_exchange/1, remove_for_exchange/1,
-         remove_for_queue/1, remove_transient_for_queue/1]).
+-export([has_for_source/1, remove_for_source/1,
+         remove_for_destination/1, remove_transient_for_destination/1]).
 
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
 
--export_type([key/0]).
+-export_type([key/0, deletions/0]).
 
 -type(key() :: binary()).
 
--type(bind_errors() :: rabbit_types:error('queue_not_found' |
-                                          'exchange_not_found' |
-                                          'exchange_and_queue_not_found')).
+-type(bind_errors() :: rabbit_types:error('source_not_found' |
+                                          'destination_not_found' |
+                                          'source_and_destination_not_found')).
 -type(bind_res() :: 'ok' | bind_errors()).
 -type(inner_fun() ::
-        fun((rabbit_types:exchange(), queue()) ->
+        fun((rabbit_types:exchange(),
+             rabbit_types:exchange() | rabbit_types:amqqueue()) ->
                    rabbit_types:ok_or_error(rabbit_types:amqp_error()))).
 -type(bindings() :: [rabbit_types:binding()]).
 
+-opaque(deletions() :: dict:dictionary()).
+
 -spec(recover/0 :: () -> [rabbit_types:binding()]).
 -spec(exists/1 :: (rabbit_types:binding()) -> boolean() | bind_errors()).
 -spec(add/1 :: (rabbit_types:binding()) -> bind_res()).
@@ -65,10 +71,13 @@
 -spec(remove/2 :: (rabbit_types:binding(), inner_fun()) ->
                        bind_res() | rabbit_types:error('binding_not_found')).
 -spec(list/1 :: (rabbit_types:vhost()) -> bindings()).
--spec(list_for_exchange/1 :: (rabbit_exchange:name()) -> bindings()).
--spec(list_for_queue/1 :: (rabbit_amqqueue:name()) -> bindings()).
--spec(list_for_exchange_and_queue/2 ::
-        (rabbit_exchange:name(), rabbit_amqqueue:name()) -> bindings()).
+-spec(list_for_source/1 ::
+        (rabbit_types:binding_source()) -> bindings()).
+-spec(list_for_destination/1 ::
+        (rabbit_types:binding_destination()) -> bindings()).
+-spec(list_for_source_and_destination/2 ::
+        (rabbit_types:binding_source(), rabbit_types:binding_destination()) ->
+                                                bindings()).
 -spec(info_keys/0 :: () -> [rabbit_types:info_key()]).
 -spec(info/1 :: (rabbit_types:binding()) -> [rabbit_types:info()]).
 -spec(info/2 :: (rabbit_types:binding(), [rabbit_types:info_key()]) ->
@@ -76,18 +85,27 @@
 -spec(info_all/1 :: (rabbit_types:vhost()) -> [[rabbit_types:info()]]).
 -spec(info_all/2 ::(rabbit_types:vhost(), [rabbit_types:info_key()])
                     -> [[rabbit_types:info()]]).
--spec(has_for_exchange/1 :: (rabbit_exchange:name()) -> boolean()).
--spec(remove_for_exchange/1 :: (rabbit_exchange:name()) -> bindings()).
--spec(remove_for_queue/1 ::
-        (rabbit_amqqueue:name()) -> fun (() -> any())).
--spec(remove_transient_for_queue/1 ::
-        (rabbit_amqqueue:name()) -> fun (() -> any())).
+-spec(has_for_source/1 :: (rabbit_types:binding_source()) -> boolean()).
+-spec(remove_for_source/1 :: (rabbit_types:binding_source()) -> bindings()).
+-spec(remove_for_destination/1 ::
+        (rabbit_types:binding_destination()) -> deletions()).
+-spec(remove_transient_for_destination/1 ::
+        (rabbit_types:binding_destination()) -> deletions()).
+-spec(process_deletions/1 :: (deletions()) -> 'ok').
+-spec(combine_deletions/2 :: (deletions(), deletions()) -> deletions()).
+-spec(add_deletion/3 :: (rabbit_exchange:name(),
+                         {'undefined' | rabbit_types:binding_source(),
+                          'deleted' | 'not_deleted',
+                          deletions()}, deletions()) -> deletions()).
+-spec(new_deletions/0 :: () -> deletions()).
 
 -endif.
 
 %%----------------------------------------------------------------------------
 
--define(INFO_KEYS, [exchange_name, queue_name, routing_key, arguments]).
+-define(INFO_KEYS, [source_name, source_kind,
+                    destination_name, destination_kind,
+                    routing_key, arguments]).
 
 recover() ->
     rabbit_misc:table_fold(
@@ -101,36 +119,34 @@ recover() ->
 exists(Binding) ->
     binding_action(
       Binding,
-      fun (_X, _Q, B) -> mnesia:read({rabbit_route, B}) /= [] end).
+      fun (_Src, _Dst, B) -> mnesia:read({rabbit_route, B}) /= [] end).
 
-add(Binding) -> add(Binding, fun (_X, _Q) -> ok end).
+add(Binding) -> add(Binding, fun (_Src, _Dst) -> ok end).
 
-remove(Binding) -> remove(Binding, fun (_X, _Q) -> ok end).
+remove(Binding) -> remove(Binding, fun (_Src, _Dst) -> ok end).
 
 add(Binding, InnerFun) ->
     case binding_action(
            Binding,
-           fun (X, Q, B) ->
+           fun (Src, Dst, B) ->
                    %% this argument is used to check queue exclusivity;
                    %% in general, we want to fail on that in preference to
                    %% anything else
-                   case InnerFun(X, Q) of
+                   case InnerFun(Src, Dst) of
                        ok ->
                            case mnesia:read({rabbit_route, B}) of
-                               []  -> Durable = (X#exchange.durable andalso
-                                                 Q#amqqueue.durable),
-                                      ok = sync_binding(
-                                             B, Durable,
+                               []  -> ok = sync_binding(
+                                             B, all_durable([Src, Dst]),
                                              fun mnesia:write/3),
-                                      {new, X, B};
-                               [_] -> {existing, X, B}
+                                      {new, Src, B};
+                               [_] -> {existing, Src, B}
                            end;
                        {error, _} = E ->
                            E
                    end
            end) of
-        {new, X = #exchange{ type = Type }, B} ->
-            ok = (type_to_module(Type)):add_binding(X, B),
+        {new, Src = #exchange{ type = Type }, B} ->
+            ok = (type_to_module(Type)):add_binding(Src, B),
             rabbit_event:notify(binding_created, info(B));
         {existing, _, _} ->
             ok;
@@ -141,61 +157,55 @@ add(Binding, InnerFun) ->
 remove(Binding, InnerFun) ->
     case binding_action(
            Binding,
-           fun (X, Q, B) ->
+           fun (Src, Dst, B) ->
                    case mnesia:match_object(rabbit_route, #route{binding = B},
                                             write) of
-                       []  -> {error, binding_not_found};
-                       [_] -> case InnerFun(X, Q) of
-                                  ok ->
-                                      Durable = (X#exchange.durable andalso
-                                                 Q#amqqueue.durable),
-                                      ok = sync_binding(
-                                             B, Durable,
-                                             fun mnesia:delete_object/3),
-                                      Deleted =
-                                          rabbit_exchange:maybe_auto_delete(X),
-                                      {{Deleted, X}, B};
-                                  {error, _} = E ->
-                                      E
-                              end
+                       [] ->
+                           {error, binding_not_found};
+                       [_] ->
+                           case InnerFun(Src, Dst) of
+                               ok ->
+                                   ok = sync_binding(
+                                          B, all_durable([Src, Dst]),
+                                          fun mnesia:delete_object/3),
+                                   {ok,
+                                    maybe_auto_delete(B#binding.source,
+                                                      [B], new_deletions())};
+                               {error, _} = E ->
+                                   E
+                           end
                    end
            end) of
         {error, _} = Err ->
             Err;
-        {{IsDeleted, X = #exchange{ type = Type }}, B} ->
-            Module = type_to_module(Type),
-            case IsDeleted of
-                auto_deleted -> ok = Module:delete(X, [B]);
-                not_deleted  -> ok = Module:remove_bindings(X, [B])
-            end,
-            rabbit_event:notify(binding_deleted, info(B)),
-            ok
+        {ok, Deletions} ->
+            ok = process_deletions(Deletions)
     end.
 
 list(VHostPath) ->
-    Route = #route{binding = #binding{
-                     exchange_name = rabbit_misc:r(VHostPath, exchange),
-                     queue_name    = rabbit_misc:r(VHostPath, queue),
-                     _             = '_'},
+    VHostResource = rabbit_misc:r(VHostPath, '_'),
+    Route = #route{binding = #binding{source      = VHostResource,
+                                      destination = VHostResource,
+                                      _           = '_'},
                    _       = '_'},
     [B || #route{binding = B} <- mnesia:dirty_match_object(rabbit_route,
                                                            Route)].
 
-list_for_exchange(XName) ->
-    Route = #route{binding = #binding{exchange_name = XName, _ = '_'}},
+list_for_source(SrcName) ->
+    Route = #route{binding = #binding{source = SrcName, _ = '_'}},
     [B || #route{binding = B} <- mnesia:dirty_match_object(rabbit_route,
                                                            Route)].
 
-list_for_queue(QueueName) ->
-    Route = #route{binding = #binding{queue_name = QueueName, _ = '_'}},
+list_for_destination(DstName) ->
+    Route = #route{binding = #binding{destination = DstName, _ = '_'}},
     [reverse_binding(B) || #reverse_route{reverse_binding = B} <-
                                mnesia:dirty_match_object(rabbit_reverse_route,
                                                          reverse_route(Route))].
 
-list_for_exchange_and_queue(XName, QueueName) ->
-    Route = #route{binding = #binding{exchange_name = XName,
-                                      queue_name    = QueueName,
-                                      _             = '_'}},
+list_for_source_and_destination(SrcName, DstName) ->
+    Route = #route{binding = #binding{source      = SrcName,
+                                      destination = DstName,
+                                      _           = '_'}},
     [B || #route{binding = B} <- mnesia:dirty_match_object(rabbit_route,
                                                            Route)].
 
@@ -208,10 +218,12 @@ map(VHostPath, F) ->
 
 infos(Items, B) -> [{Item, i(Item, B)} || Item <- Items].
 
-i(exchange_name, #binding{exchange_name = XName})      -> XName;
-i(queue_name,    #binding{queue_name    = QName})      -> QName;
-i(routing_key,   #binding{key           = RoutingKey}) -> RoutingKey;
-i(arguments,     #binding{args          = Arguments})  -> Arguments;
+i(source_name,      #binding{source      = SrcName})    -> SrcName#resource.name;
+i(source_kind,      #binding{source      = SrcName})    -> SrcName#resource.kind;
+i(destination_name, #binding{destination = DstName})    -> DstName#resource.name;
+i(destination_kind, #binding{destination = DstName})    -> DstName#resource.kind;
+i(routing_key,      #binding{key         = RoutingKey}) -> RoutingKey;
+i(arguments,        #binding{args        = Arguments})  -> Arguments;
 i(Item, _) -> throw({bad_argument, Item}).
 
 info(B = #binding{}) -> infos(?INFO_KEYS, B).
@@ -222,14 +234,14 @@ info_all(VHostPath) -> map(VHostPath, fun (B) -> info(B) end).
 
 info_all(VHostPath, Items) -> map(VHostPath, fun (B) -> info(B, Items) end).
 
-has_for_exchange(XName) ->
-    Match = #route{binding = #binding{exchange_name = XName, _ = '_'}},
+has_for_source(SrcName) ->
+    Match = #route{binding = #binding{source = SrcName, _ = '_'}},
     %% we need to check for durable routes here too in case a bunch of
     %% routes to durable queues have been removed temporarily as a
     %% result of a node failure
     contains(rabbit_route, Match) orelse contains(rabbit_durable_route, Match).
 
-remove_for_exchange(XName) ->
+remove_for_source(SrcName) ->
     [begin
          ok = mnesia:delete_object(rabbit_reverse_route,
                                    reverse_route(Route), write),
@@ -237,26 +249,31 @@ remove_for_exchange(XName) ->
          Route#route.binding
      end || Route <- mnesia:match_object(
                        rabbit_route,
-                       #route{binding = #binding{exchange_name = XName,
-                                                 _ = '_'}},
+                       #route{binding = #binding{source = SrcName,
+                                                 _      = '_'}},
                        write)].
 
-remove_for_queue(QueueName) ->
-    remove_for_queue(QueueName, fun delete_forward_routes/1).
+remove_for_destination(DstName) ->
+    remove_for_destination(DstName, fun delete_forward_routes/1).
 
-remove_transient_for_queue(QueueName) ->
-    remove_for_queue(QueueName, fun delete_transient_forward_routes/1).
+remove_transient_for_destination(DstName) ->
+    remove_for_destination(DstName, fun delete_transient_forward_routes/1).
 
 %%----------------------------------------------------------------------------
 
-binding_action(Binding = #binding{exchange_name = XName,
-                                  queue_name    = QueueName,
-                                  args          = Arguments}, Fun) ->
-    call_with_exchange_and_queue(
-      XName, QueueName,
-      fun (X, Q) ->
+all_durable(Resources) ->
+    lists:all(fun (#exchange{durable = D}) -> D;
+                  (#amqqueue{durable = D}) -> D
+              end, Resources).
+
+binding_action(Binding = #binding{source      = SrcName,
+                                  destination = DstName,
+                                  args        = Arguments}, Fun) ->
+    call_with_source_and_destination(
+      SrcName, DstName,
+      fun (Src, Dst) ->
               SortedArgs = rabbit_misc:sort_field_table(Arguments),
-              Fun(X, Q, Binding#binding{args = SortedArgs})
+              Fun(Src, Dst, Binding#binding{args = SortedArgs})
       end).
 
 sync_binding(Binding, Durable, Fun) ->
@@ -270,17 +287,22 @@ sync_binding(Binding, Durable, Fun) ->
     ok = Fun(rabbit_reverse_route, ReverseRoute, write),
     ok.
 
-call_with_exchange_and_queue(XName, QueueName, Fun) ->
+call_with_source_and_destination(SrcName, DstName, Fun) ->
+    SrcTable = table_for_resource(SrcName),
+    DstTable = table_for_resource(DstName),
     rabbit_misc:execute_mnesia_transaction(
-      fun () -> case {mnesia:read({rabbit_exchange, XName}),
-                      mnesia:read({rabbit_queue, QueueName})} of
-                   {[X], [Q]} -> Fun(X, Q);
-                   {[ ], [_]} -> {error, exchange_not_found};
-                   {[_], [ ]} -> {error, queue_not_found};
-                   {[ ], [ ]} -> {error, exchange_and_queue_not_found}
-               end
+      fun () -> case {mnesia:read({SrcTable, SrcName}),
+                      mnesia:read({DstTable, DstName})} of
+                    {[Src], [Dst]} -> Fun(Src, Dst);
+                    {[],    [_]  } -> {error, source_not_found};
+                    {[_],   []   } -> {error, destination_not_found};
+                    {[],    []   } -> {error, source_and_destination_not_found}
+                end
       end).
 
+table_for_resource(#resource{kind = exchange}) -> rabbit_exchange;
+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),
@@ -293,8 +315,8 @@ continue('$end_of_table')    -> false;
 continue({[_|_], _})         -> true;
 continue({[], Continuation}) -> continue(mnesia:select(Continuation)).
 
-remove_for_queue(QueueName, FwdDeleteFun) ->
-    DeletedBindings =
+remove_for_destination(DstName, FwdDeleteFun) ->
+    Bindings =
         [begin
              Route = reverse_route(ReverseRoute),
              ok = FwdDeleteFun(Route),
@@ -304,40 +326,41 @@ remove_for_queue(QueueName, FwdDeleteFun) ->
          end || ReverseRoute
                     <- mnesia:match_object(
                          rabbit_reverse_route,
-                         reverse_route(#route{binding = #binding{
-                                                queue_name = QueueName,
-                                                _          = '_'}}),
+                         reverse_route(#route{
+                                          binding = #binding{
+                                            destination = DstName,
+                                            _           = '_'}}),
                          write)],
-    Grouped = group_bindings_and_auto_delete(
-                lists:keysort(#binding.exchange_name, DeletedBindings), []),
-    fun () ->
-            lists:foreach(
-              fun ({{IsDeleted, X = #exchange{ type = Type }}, Bs}) ->
-                      Module = type_to_module(Type),
-                      case IsDeleted of
-                          auto_deleted -> Module:delete(X, Bs);
-                          not_deleted  -> Module:remove_bindings(X, Bs)
-                      end
-              end, Grouped)
-    end.
+    group_bindings_fold(fun maybe_auto_delete/3, new_deletions(),
+                        lists:keysort(#binding.source, Bindings)).
 
 %% Requires that its input binding list is sorted in exchange-name
 %% order, so that the grouping of bindings (for passing to
 %% group_bindings_and_auto_delete1) works properly.
-group_bindings_and_auto_delete([], Acc) ->
+group_bindings_fold(_Fun, Acc, []) ->
     Acc;
-group_bindings_and_auto_delete(
-  [B = #binding{exchange_name = XName} | Bs], Acc) ->
-    group_bindings_and_auto_delete(XName, Bs, [B], Acc).
-
-group_bindings_and_auto_delete(
-  XName, [B = #binding{exchange_name = XName} | Bs], Bindings, Acc) ->
-    group_bindings_and_auto_delete(XName, Bs, [B | Bindings], Acc);
-group_bindings_and_auto_delete(XName, Removed, Bindings, Acc) ->
-    %% either Removed is [], or its head has a non-matching XName
-    [X] = mnesia:read({rabbit_exchange, XName}),
-    NewAcc = [{{rabbit_exchange:maybe_auto_delete(X), X}, Bindings} | Acc],
-    group_bindings_and_auto_delete(Removed, NewAcc).
+group_bindings_fold(Fun, Acc, [B = #binding{source = SrcName} | Bs]) ->
+    group_bindings_fold(Fun, SrcName, Acc, Bs, [B]).
+
+group_bindings_fold(
+  Fun, SrcName, Acc, [B = #binding{source = SrcName} | Bs], Bindings) ->
+    group_bindings_fold(Fun, SrcName, Acc, Bs, [B | Bindings]);
+group_bindings_fold(Fun, SrcName, Acc, Removed, Bindings) ->
+    %% Either Removed is [], or its head has a non-matching SrcName.
+    group_bindings_fold(Fun, Fun(SrcName, Bindings, Acc), Removed).
+
+maybe_auto_delete(XName, Bindings, Deletions) ->
+    case rabbit_exchange:lookup(XName) of
+        {error, not_found} ->
+            add_deletion(XName, {undefined, not_deleted, Bindings}, Deletions);
+        {ok, X} ->
+            add_deletion(XName, {X, not_deleted, Bindings},
+                         case rabbit_exchange:maybe_auto_delete(X) of
+                             not_deleted           -> Deletions;
+                             {deleted, Deletions1} -> combine_deletions(
+                                                        Deletions, Deletions1)
+                         end)
+    end.
 
 delete_forward_routes(Route) ->
     ok = mnesia:delete_object(rabbit_route, Route, write),
@@ -358,20 +381,59 @@ reverse_route(#route{binding = Binding}) ->
 reverse_route(#reverse_route{reverse_binding = Binding}) ->
     #route{binding = reverse_binding(Binding)}.
 
-reverse_binding(#reverse_binding{exchange_name = XName,
-                                 queue_name    = QueueName,
-                                 key           = Key,
-                                 args          = Args}) ->
-    #binding{exchange_name = XName,
-             queue_name    = QueueName,
-             key           = Key,
-             args          = Args};
-
-reverse_binding(#binding{exchange_name = XName,
-                         queue_name    = QueueName,
-                         key           = Key,
-                         args          = Args}) ->
-    #reverse_binding{exchange_name = XName,
-                     queue_name    = QueueName,
-                     key           = Key,
-                     args          = Args}.
+reverse_binding(#reverse_binding{source      = SrcName,
+                                 destination = DstName,
+                                 key         = Key,
+                                 args        = Args}) ->
+    #binding{source      = SrcName,
+             destination = DstName,
+             key         = Key,
+             args        = Args};
+
+reverse_binding(#binding{source      = SrcName,
+                         destination = DstName,
+                         key         = Key,
+                         args        = Args}) ->
+    #reverse_binding{source      = SrcName,
+                     destination = DstName,
+                     key         = Key,
+                     args        = Args}.
+
+%% ----------------------------------------------------------------------------
+%% Binding / exchange deletion abstraction API
+%% ----------------------------------------------------------------------------
+
+anything_but( NotThis, NotThis, NotThis) -> NotThis;
+anything_but( NotThis, NotThis,    This) -> This;
+anything_but( NotThis,    This, NotThis) -> This;
+anything_but(_NotThis,    This,    This) -> This.
+
+new_deletions() -> dict:new().
+
+add_deletion(XName, Entry, Deletions) ->
+    dict:update(XName, fun (Entry1) -> merge_entry(Entry1, Entry) end,
+                Entry, Deletions).
+
+combine_deletions(Deletions1, Deletions2) ->
+    dict:merge(fun (_XName, Entry1, Entry2) -> merge_entry(Entry1, Entry2) end,
+               Deletions1, Deletions2).
+
+merge_entry({X1, Deleted1, Bindings1}, {X2, Deleted2, Bindings2}) ->
+    {anything_but(undefined, X1, X2),
+     anything_but(not_deleted, Deleted1, Deleted2),
+     [Bindings1 | Bindings2]}.
+
+process_deletions(Deletions) ->
+    dict:fold(
+      fun (_XName, {X = #exchange{ type = Type }, Deleted, Bindings}, ok) ->
+              FlatBindings = lists:flatten(Bindings),
+              [rabbit_event:notify(binding_deleted, info(B)) ||
+                  B <- FlatBindings],
+              TypeModule = type_to_module(Type),
+              case Deleted of
+                  not_deleted -> TypeModule:remove_bindings(X, FlatBindings);
+                  deleted     -> rabbit_event:notify(exchange_deleted,
+                                                     [{name, X#exchange.name}]),
+                                 TypeModule:delete(X, FlatBindings)
+              end
+      end, ok, Deletions).
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index bde11f00e0..58c8e34122 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -58,7 +58,8 @@
          consumer_count,
          messages_unacknowledged,
          acks_uncommitted,
-         prefetch_count]).
+         prefetch_count,
+         client_flow_blocked]).
 
 -define(CREATION_EVENT_KEYS,
         [pid,
@@ -314,14 +315,10 @@ terminating(Reason, State = #ch{channel = Channel, reader_pid = Reader}) ->
 
 return_queue_declare_ok(#resource{name = ActualName},
                         NoWait, MessageCount, ConsumerCount, State) ->
-    NewState = State#ch{most_recently_declared_queue = ActualName},
-    case NoWait of
-        true  -> {noreply, NewState};
-        false -> Reply = #'queue.declare_ok'{queue = ActualName,
-                                             message_count = MessageCount,
-                                             consumer_count = ConsumerCount},
-                 {reply, Reply, NewState}
-    end.
+    return_ok(State#ch{most_recently_declared_queue = ActualName}, NoWait,
+              #'queue.declare_ok'{queue          = ActualName,
+                                  message_count  = MessageCount,
+                                  consumer_count = ConsumerCount}).
 
 check_resource_access(Username, Resource, Perm) ->
     V = {Resource, Perm},
@@ -343,34 +340,47 @@ clear_permission_cache() ->
     erase(permission_cache),
     ok.
 
-check_configure_permitted(Resource, #ch{ username = Username}) ->
+check_configure_permitted(Resource, #ch{username = Username}) ->
     check_resource_access(Username, Resource, configure).
 
-check_write_permitted(Resource, #ch{ username = Username}) ->
+check_write_permitted(Resource, #ch{username = Username}) ->
     check_resource_access(Username, Resource, write).
 
-check_read_permitted(Resource, #ch{ username = Username}) ->
+check_read_permitted(Resource, #ch{username = Username}) ->
     check_resource_access(Username, Resource, read).
 
-expand_queue_name_shortcut(<<>>, #ch{ most_recently_declared_queue = <<>> }) ->
+expand_queue_name_shortcut(<<>>, #ch{most_recently_declared_queue = <<>>}) ->
     rabbit_misc:protocol_error(
       not_found, "no previously declared queue", []);
-expand_queue_name_shortcut(<<>>, #ch{ virtual_host = VHostPath,
-                                      most_recently_declared_queue = MRDQ }) ->
+expand_queue_name_shortcut(<<>>, #ch{virtual_host = VHostPath,
+                                     most_recently_declared_queue = MRDQ}) ->
     rabbit_misc:r(VHostPath, queue, MRDQ);
-expand_queue_name_shortcut(QueueNameBin, #ch{ virtual_host = VHostPath }) ->
+expand_queue_name_shortcut(QueueNameBin, #ch{virtual_host = VHostPath}) ->
     rabbit_misc:r(VHostPath, queue, QueueNameBin).
 
 expand_routing_key_shortcut(<<>>, <<>>,
-                            #ch{ most_recently_declared_queue = <<>> }) ->
+                            #ch{most_recently_declared_queue = <<>>}) ->
     rabbit_misc:protocol_error(
       not_found, "no previously declared queue", []);
 expand_routing_key_shortcut(<<>>, <<>>,
-                            #ch{ most_recently_declared_queue = MRDQ }) ->
+                            #ch{most_recently_declared_queue = MRDQ}) ->
     MRDQ;
 expand_routing_key_shortcut(_QueueNameBin, RoutingKey, _State) ->
     RoutingKey.
 
+expand_binding(queue, DestinationNameBin, RoutingKey, State) ->
+    {expand_queue_name_shortcut(DestinationNameBin, State),
+     expand_routing_key_shortcut(DestinationNameBin, RoutingKey, State)};
+expand_binding(exchange, DestinationNameBin, RoutingKey, State) ->
+    {rabbit_misc:r(State#ch.virtual_host, exchange, DestinationNameBin),
+     RoutingKey}.
+
+check_not_default_exchange(#resource{kind = exchange, name = <<"">>}) ->
+    rabbit_misc:protocol_error(
+      access_refused, "operation not permitted on the default exchange", []);
+check_not_default_exchange(_) ->
+    ok.
+
 %% check that an exchange/queue name does not contain the reserved
 %% "amq."  prefix.
 %%
@@ -437,11 +447,11 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     %% certain to want to look at delivery-mode and priority.
     DecodedContent = rabbit_binary_parser:ensure_content_decoded(Content),
     IsPersistent = is_message_persistent(DecodedContent),
-    Message = #basic_message{exchange_name  = ExchangeName,
-                             routing_key    = RoutingKey,
-                             content        = DecodedContent,
-                             guid           = rabbit_guid:guid(),
-                             is_persistent  = IsPersistent},
+    Message = #basic_message{exchange_name = ExchangeName,
+                             routing_key   = RoutingKey,
+                             content       = DecodedContent,
+                             guid          = rabbit_guid:guid(),
+                             is_persistent = IsPersistent},
     {RoutingRes, DeliveredQPids} =
         rabbit_exchange:publish(
           Exchange,
@@ -480,9 +490,9 @@ handle_method(#'basic.ack'{delivery_tag = DeliveryTag,
 
 handle_method(#'basic.get'{queue = QueueNameBin,
                            no_ack = NoAck},
-              _, State = #ch{ writer_pid = WriterPid,
-                              reader_pid = ReaderPid,
-                              next_tag = DeliveryTag }) ->
+              _, State = #ch{writer_pid = WriterPid,
+                             reader_pid = ReaderPid,
+                             next_tag   = DeliveryTag}) ->
     QueueName = expand_queue_name_shortcut(QueueNameBin, State),
     check_read_permitted(QueueName, State),
     case rabbit_amqqueue:with_exclusive_access_or_die(
@@ -512,15 +522,15 @@ handle_method(#'basic.get'{queue = QueueNameBin,
             {reply, #'basic.get_empty'{}, State}
     end;
 
-handle_method(#'basic.consume'{queue = QueueNameBin,
+handle_method(#'basic.consume'{queue        = QueueNameBin,
                                consumer_tag = ConsumerTag,
-                               no_local = _, % FIXME: implement
-                               no_ack = NoAck,
-                               exclusive = ExclusiveConsume,
-                               nowait = NoWait},
-              _, State = #ch{ reader_pid = ReaderPid,
-                              limiter_pid = LimiterPid,
-                              consumer_mapping = ConsumerMapping }) ->
+                               no_local     = _, % FIXME: implement
+                               no_ack       = NoAck,
+                               exclusive    = ExclusiveConsume,
+                               nowait       = NoWait},
+              _, State = #ch{reader_pid       = ReaderPid,
+                             limiter_pid      = LimiterPid,
+                             consumer_mapping = ConsumerMapping }) ->
     case dict:find(ConsumerTag, ConsumerMapping) of
         error ->
             QueueName = expand_queue_name_shortcut(QueueNameBin, State),
@@ -615,7 +625,7 @@ handle_method(#'basic.qos'{prefetch_count = PrefetchCount},
     {reply, #'basic.qos_ok'{}, State#ch{limiter_pid = LimiterPid2}};
 
 handle_method(#'basic.recover_async'{requeue = true},
-              _, State = #ch{ unacked_message_q = UAMQ }) ->
+              _, State = #ch{unacked_message_q = UAMQ}) ->
     ok = fold_per_queue(
            fun (QPid, MsgIds, ok) ->
                    %% The Qpid python test suite incorrectly assumes
@@ -630,8 +640,8 @@ handle_method(#'basic.recover_async'{requeue = true},
     {noreply, State#ch{unacked_message_q = queue:new()}};
 
 handle_method(#'basic.recover_async'{requeue = false},
-              _, State = #ch{ writer_pid = WriterPid,
-                              unacked_message_q = UAMQ }) ->
+              _, State = #ch{writer_pid = WriterPid,
+                             unacked_message_q = UAMQ}) ->
     ok = rabbit_misc:queue_fold(
            fun ({_DeliveryTag, none, _Msg}, ok) ->
                    %% Was sent as a basic.get_ok. Don't redeliver
@@ -664,7 +674,7 @@ handle_method(#'basic.recover'{requeue = Requeue}, Content, State) ->
 
 handle_method(#'basic.reject'{delivery_tag = DeliveryTag,
                               requeue = Requeue},
-              _, State = #ch{ unacked_message_q = UAMQ}) ->
+              _, State = #ch{unacked_message_q = UAMQ}) ->
     {Acked, Remaining} = collect_acks(UAMQ, DeliveryTag, false),
     ok = fold_per_queue(
            fun (QPid, MsgIds, ok) ->
@@ -681,9 +691,10 @@ handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
                                   internal = false,
                                   nowait = NoWait,
                                   arguments = Args},
-              _, State = #ch{ virtual_host = VHostPath }) ->
+              _, State = #ch{virtual_host = VHostPath}) ->
     CheckedType = rabbit_exchange:check_type(TypeNameBin),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
+    check_not_default_exchange(ExchangeName),
     check_configure_permitted(ExchangeName, State),
     X = case rabbit_exchange:lookup(ExchangeName) of
             {ok, FoundX} -> FoundX;
@@ -709,17 +720,19 @@ handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
 handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
                                   passive = true,
                                   nowait = NoWait},
-              _, State = #ch{ virtual_host = VHostPath }) ->
+              _, State = #ch{virtual_host = VHostPath}) ->
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_configure_permitted(ExchangeName, State),
+    check_not_default_exchange(ExchangeName),
     _ = rabbit_exchange:lookup_or_die(ExchangeName),
     return_ok(State, NoWait, #'exchange.declare_ok'{});
 
 handle_method(#'exchange.delete'{exchange = ExchangeNameBin,
                                  if_unused = IfUnused,
                                  nowait = NoWait},
-              _, State = #ch { virtual_host = VHostPath }) ->
+              _, State = #ch{virtual_host = VHostPath}) ->
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
+    check_not_default_exchange(ExchangeName),
     check_configure_permitted(ExchangeName, State),
     case rabbit_exchange:delete(ExchangeName, IfUnused) of
         {error, not_found} ->
@@ -731,6 +744,24 @@ handle_method(#'exchange.delete'{exchange = ExchangeNameBin,
             return_ok(State, NoWait,  #'exchange.delete_ok'{})
     end;
 
+handle_method(#'exchange.bind'{destination = DestinationNameBin,
+                               source = SourceNameBin,
+                               routing_key = RoutingKey,
+                               nowait = NoWait,
+                               arguments = Arguments}, _, State) ->
+    binding_action(fun rabbit_binding:add/2,
+                   SourceNameBin, exchange, DestinationNameBin, RoutingKey,
+                   Arguments, #'exchange.bind_ok'{}, NoWait, State);
+
+handle_method(#'exchange.unbind'{destination = DestinationNameBin,
+                                 source = SourceNameBin,
+                                 routing_key = RoutingKey,
+                                 nowait = NoWait,
+                                 arguments = Arguments}, _, State) ->
+    binding_action(fun rabbit_binding:remove/2,
+                   SourceNameBin, exchange, DestinationNameBin, RoutingKey,
+                   Arguments, #'exchange.unbind_ok'{}, NoWait, State);
+
 handle_method(#'queue.declare'{queue       = QueueNameBin,
                                passive     = false,
                                durable     = Durable,
@@ -822,7 +853,7 @@ handle_method(#'queue.bind'{queue = QueueNameBin,
                             nowait = NoWait,
                             arguments = Arguments}, _, State) ->
     binding_action(fun rabbit_binding:add/2,
-                   ExchangeNameBin, QueueNameBin, RoutingKey, Arguments,
+                   ExchangeNameBin, queue, QueueNameBin, RoutingKey, Arguments,
                    #'queue.bind_ok'{}, NoWait, State);
 
 handle_method(#'queue.unbind'{queue = QueueNameBin,
@@ -830,7 +861,7 @@ handle_method(#'queue.unbind'{queue = QueueNameBin,
                               routing_key = RoutingKey,
                               arguments = Arguments}, _, State) ->
     binding_action(fun rabbit_binding:remove/2,
-                   ExchangeNameBin, QueueNameBin, RoutingKey, Arguments,
+                   ExchangeNameBin, queue, QueueNameBin, RoutingKey, Arguments,
                    #'queue.unbind_ok'{}, false, State);
 
 handle_method(#'queue.purge'{queue = QueueNameBin,
@@ -872,6 +903,7 @@ handle_method(#'channel.flow'{active = true}, _,
                   end,
     {reply, #'channel.flow_ok'{active = true},
      State#ch{limiter_pid = LimiterPid1}};
+
 handle_method(#'channel.flow'{active = false}, _,
               State = #ch{limiter_pid = LimiterPid,
                           consumer_mapping = Consumers}) ->
@@ -879,14 +911,14 @@ handle_method(#'channel.flow'{active = false}, _,
                       undefined -> start_limiter(State);
                       Other     -> Other
                   end,
+    State1 = State#ch{limiter_pid = LimiterPid1},
     ok = rabbit_limiter:block(LimiterPid1),
-    QPids = consumer_queues(Consumers),
-    Queues = [{QPid, erlang:monitor(process, QPid)} || QPid <- QPids],
-    ok = rabbit_amqqueue:flush_all(QPids, self()),
-    case Queues of
-        [] -> {reply, #'channel.flow_ok'{active = false}, State};
-        _  -> {noreply, State#ch{limiter_pid = LimiterPid1,
-                                 blocking = dict:from_list(Queues)}}
+    case consumer_queues(Consumers) of
+        []    -> {reply, #'channel.flow_ok'{active = false}, State1};
+        QPids -> Queues = [{QPid, erlang:monitor(process, QPid)} ||
+                              QPid <- QPids],
+                 ok = rabbit_amqqueue:flush_all(QPids, self()),
+                 {noreply, State1#ch{blocking = dict:from_list(Queues)}}
     end;
 
 handle_method(_MethodRecord, _Content, _State) ->
@@ -895,42 +927,44 @@ handle_method(_MethodRecord, _Content, _State) ->
 
 %%----------------------------------------------------------------------------
 
-binding_action(Fun, ExchangeNameBin, QueueNameBin, RoutingKey, Arguments,
-               ReturnMethod, NoWait,
+binding_action(Fun, ExchangeNameBin, DestinationType, DestinationNameBin,
+               RoutingKey, Arguments, ReturnMethod, NoWait,
                State = #ch{virtual_host = VHostPath,
                            reader_pid   = ReaderPid}) ->
     %% FIXME: connection exception (!) on failure??
     %% (see rule named "failure" in spec-XML)
     %% FIXME: don't allow binding to internal exchanges -
     %% including the one named "" !
-    QueueName = expand_queue_name_shortcut(QueueNameBin, State),
-    check_write_permitted(QueueName, State),
-    ActualRoutingKey = expand_routing_key_shortcut(QueueNameBin, RoutingKey,
-                                                   State),
+    {DestinationName, ActualRoutingKey} =
+        expand_binding(DestinationType, DestinationNameBin, RoutingKey, State),
+    check_write_permitted(DestinationName, State),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
+    [check_not_default_exchange(N) || N <- [DestinationName, ExchangeName]],
     check_read_permitted(ExchangeName, State),
-    case Fun(#binding{exchange_name = ExchangeName,
-                      queue_name    = QueueName,
-                      key           = ActualRoutingKey,
-                      args          = Arguments},
-             fun (_X, Q) ->
+    case Fun(#binding{source      = ExchangeName,
+                      destination = DestinationName,
+                      key         = ActualRoutingKey,
+                      args        = Arguments},
+             fun (_X, Q = #amqqueue{}) ->
                      try rabbit_amqqueue:check_exclusive_access(Q, ReaderPid)
                      catch exit:Reason -> {error, Reason}
-                     end
+                     end;
+                 (_X, #exchange{}) ->
+                     ok
              end) of
-        {error, exchange_not_found} ->
+        {error, source_not_found} ->
             rabbit_misc:not_found(ExchangeName);
-        {error, queue_not_found} ->
-            rabbit_misc:not_found(QueueName);
-        {error, exchange_and_queue_not_found} ->
+        {error, destination_not_found} ->
+            rabbit_misc:not_found(DestinationName);
+        {error, source_and_destination_not_found} ->
             rabbit_misc:protocol_error(
               not_found, "no ~s and no ~s", [rabbit_misc:rs(ExchangeName),
-                                             rabbit_misc:rs(QueueName)]);
+                                             rabbit_misc:rs(DestinationName)]);
         {error, binding_not_found} ->
             rabbit_misc:protocol_error(
               not_found, "no binding ~s between ~s and ~s",
               [RoutingKey, rabbit_misc:rs(ExchangeName),
-               rabbit_misc:rs(QueueName)]);
+               rabbit_misc:rs(DestinationName)]);
         {error, #amqp_error{} = Error} ->
             rabbit_misc:protocol_error(Error);
         ok -> return_ok(State, NoWait, ReturnMethod)
@@ -1127,6 +1161,8 @@ i(acks_uncommitted, #ch{uncommitted_ack_q = UAQ}) ->
     queue:len(UAQ);
 i(prefetch_count, #ch{limiter_pid = LimiterPid}) ->
     rabbit_limiter:get_limit(LimiterPid);
+i(client_flow_blocked, #ch{limiter_pid = LimiterPid}) ->
+    rabbit_limiter:is_blocked(LimiterPid);
 i(Item, _) ->
     throw({bad_argument, Item}).
 
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index a3b6f369e3..8facaf1606 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -94,29 +94,29 @@ start() ->
             end,
             halt();
         {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
-            error("invalid command '~s'",
-                  [lists:flatten(
-                     rabbit_misc:intersperse(
-                       " ", [atom_to_list(Command) | Args]))]),
+            print_error("invalid command '~s'",
+                        [lists:flatten(
+                           rabbit_misc:intersperse(
+                             " ", [atom_to_list(Command) | Args]))]),
             usage();
         {error, Reason} ->
-            error("~p", [Reason]),
+            print_error("~p", [Reason]),
             halt(2);
         {badrpc, {'EXIT', Reason}} ->
-            error("~p", [Reason]),
+            print_error("~p", [Reason]),
             halt(2);
         {badrpc, Reason} ->
-            error("unable to connect to node ~w: ~w", [Node, Reason]),
+            print_error("unable to connect to node ~w: ~w", [Node, Reason]),
             print_badrpc_diagnostics(Node),
             halt(2);
         Other ->
-            error("~p", [Other]),
+            print_error("~p", [Other]),
             halt(2)
     end.
 
 fmt_stderr(Format, Args) -> rabbit_misc:format_stderr(Format ++ "~n", Args).
 
-error(Format, Args) -> fmt_stderr("Error: " ++ Format, Args).
+print_error(Format, Args) -> fmt_stderr("Error: " ++ Format, Args).
 
 print_badrpc_diagnostics(Node) ->
     fmt_stderr("diagnostics:", []),
@@ -257,7 +257,8 @@ action(list_exchanges, Node, Args, Opts, Inform) ->
 action(list_bindings, Node, Args, Opts, Inform) ->
     Inform("Listing bindings", []),
     VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
-    ArgAtoms = default_if_empty(Args, [exchange_name, queue_name,
+    ArgAtoms = default_if_empty(Args, [source_name, source_kind,
+                                       destination_name, destination_kind,
                                        routing_key, arguments]),
     display_info_list(rpc_call(Node, rabbit_binding, info_all,
                                [VHostArg, ArgAtoms]),
@@ -347,6 +348,8 @@ format_info_item([{TableEntryKey, TableEntryType, _TableEntryValue} | _] =
                      Value) when is_binary(TableEntryKey) andalso
                                  is_atom(TableEntryType) ->
     io_lib:format("~1000000000000p", [prettify_amqp_table(Value)]);
+format_info_item([C|_] = Value) when is_number(C), C >= 32, C =< 255 ->
+    Value;
 format_info_item(Value) ->
     io_lib:format("~w", [Value]).
 
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 2a19d5b1c8..465642332c 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -82,8 +82,9 @@
         (name(), boolean())-> 'ok' |
                               rabbit_types:error('not_found') |
                               rabbit_types:error('in_use')).
--spec(maybe_auto_delete/1:: (rabbit_types:exchange()) ->
-                                 'not_deleted' | 'auto_deleted').
+-spec(maybe_auto_delete/1::
+        (rabbit_types:exchange())
+        -> 'not_deleted' | {'deleted', rabbit_binding:deletions()}).
 
 -endif.
 
@@ -99,11 +100,11 @@ recover() ->
            end, [], rabbit_durable_exchange),
     Bs = rabbit_binding:recover(),
     recover_with_bindings(
-      lists:keysort(#binding.exchange_name, Bs),
+      lists:keysort(#binding.source, Bs),
       lists:keysort(#exchange.name, Xs), []).
 
-recover_with_bindings([B = #binding{exchange_name = Name} | Rest],
-                      Xs = [#exchange{name = Name} | _],
+recover_with_bindings([B = #binding{source = XName} | Rest],
+                      Xs = [#exchange{name = XName} | _],
                       Bindings) ->
     recover_with_bindings(Rest, Xs, [B | Bindings]);
 recover_with_bindings(Bs, [X = #exchange{type = Type} | Xs], Bindings) ->
@@ -225,38 +226,44 @@ info_all(VHostPath) -> map(VHostPath, fun (X) -> info(X) end).
 
 info_all(VHostPath, Items) -> map(VHostPath, fun (X) -> info(X, Items) end).
 
-publish(X, Delivery) ->
-    publish(X, [], Delivery).
-
-publish(X = #exchange{type = Type}, Seen, Delivery) ->
-    case (type_to_module(Type)):publish(X, Delivery) of
-        {_, []} = R ->
-            #exchange{name = XName, arguments = Args} = X,
-            case rabbit_misc:r_arg(XName, exchange, Args,
-                                   <<"alternate-exchange">>) of
-                undefined ->
-                    R;
-                AName ->
-                    NewSeen = [XName | Seen],
-                    case lists:member(AName, NewSeen) of
-                        true  -> R;
-                        false -> case lookup(AName) of
-                                     {ok, AX} ->
-                                         publish(AX, NewSeen, Delivery);
-                                     {error, not_found} ->
-                                         rabbit_log:warning(
-                                           "alternate exchange for ~s "
-                                           "does not exist: ~s",
-                                           [rabbit_misc:rs(XName),
-                                            rabbit_misc:rs(AName)]),
-                                         R
-                                 end
-                    end
-            end;
-        R ->
-            R
+publish(X = #exchange{name = XName}, Delivery) ->
+    rabbit_router:deliver(
+      route(Delivery, {queue:from_list([X]), sets:from_list([XName]), []}),
+      Delivery).
+
+route(Delivery, {WorkList, SeenXs, QNames}) ->
+    case queue:out(WorkList) of
+        {empty, _WorkList} ->
+            lists:usort(QNames);
+        {{value, X = #exchange{type = Type}}, WorkList1} ->
+            DstNames = process_alternate(
+                         X, ((type_to_module(Type)):route(X, Delivery))),
+            route(Delivery,
+                  lists:foldl(fun process_route/2, {WorkList1, SeenXs, QNames},
+                              DstNames))
     end.
 
+process_alternate(#exchange{name = XName, arguments = Args}, []) ->
+    case rabbit_misc:r_arg(XName, exchange, Args, <<"alternate-exchange">>) of
+        undefined -> [];
+        AName     -> [AName]
+    end;
+process_alternate(_X, Results) ->
+    Results.
+
+process_route(#resource{kind = exchange} = XName,
+              {WorkList, SeenXs, QNames} = Acc) ->
+    case sets:is_element(XName, SeenXs) of
+        true  -> Acc;
+        false -> {case lookup(XName) of
+                      {ok, X}            -> queue:in(X, WorkList);
+                      {error, not_found} -> WorkList
+                  end, sets:add_element(XName, SeenXs), QNames}
+    end;
+process_route(#resource{kind = queue} = QName,
+              {WorkList, SeenXs, QNames}) ->
+    {WorkList, SeenXs, [QName | QNames]}.
+
 call_with_exchange(XName, Fun) ->
     rabbit_misc:execute_mnesia_transaction(
       fun () -> case mnesia:read({rabbit_exchange, XName}) of
@@ -271,9 +278,10 @@ delete(XName, IfUnused) ->
               false -> fun unconditional_delete/1
           end,
     case call_with_exchange(XName, Fun) of
-        {deleted, X = #exchange{type = Type}, Bs} ->
-            (type_to_module(Type)):delete(X, Bs),
-            ok;
+        {deleted, X, Bs, Deletions} ->
+            ok = rabbit_binding:process_deletions(
+                   rabbit_binding:add_deletion(
+                     XName, {X, deleted, Bs}, Deletions));
         Error = {error, _InUseOrNotFound} ->
             Error
     end.
@@ -282,19 +290,18 @@ maybe_auto_delete(#exchange{auto_delete = false}) ->
     not_deleted;
 maybe_auto_delete(#exchange{auto_delete = true} = X) ->
     case conditional_delete(X) of
-        {error, in_use}  -> not_deleted;
-        {deleted, X, []} -> auto_deleted
+        {error, in_use}             -> not_deleted;
+        {deleted, X, [], Deletions} -> {deleted, Deletions}
     end.
 
 conditional_delete(X = #exchange{name = XName}) ->
-    case rabbit_binding:has_for_exchange(XName) of
+    case rabbit_binding:has_for_source(XName) of
         false  -> unconditional_delete(X);
         true   -> {error, in_use}
     end.
 
 unconditional_delete(X = #exchange{name = XName}) ->
-    Bindings = rabbit_binding:remove_for_exchange(XName),
     ok = mnesia:delete({rabbit_durable_exchange, XName}),
     ok = mnesia:delete({rabbit_exchange, XName}),
-    rabbit_event:notify(exchange_deleted, [{name, XName}]),
-    {deleted, X, Bindings}.
+    Bindings = rabbit_binding:remove_for_source(XName),
+    {deleted, X, Bindings, rabbit_binding:remove_for_destination(XName)}.
diff --git a/src/rabbit_exchange_type.erl b/src/rabbit_exchange_type.erl
index 85760edce4..742944dcd5 100644
--- a/src/rabbit_exchange_type.erl
+++ b/src/rabbit_exchange_type.erl
@@ -36,7 +36,7 @@
 behaviour_info(callbacks) ->
     [
      {description, 0},
-     {publish, 2},
+     {route, 2},
 
      %% called BEFORE declaration, to check args etc; may exit with #amqp_error{}
      {validate, 1},
diff --git a/src/rabbit_exchange_type_direct.erl b/src/rabbit_exchange_type_direct.erl
index 4f6eb85199..d934a49709 100644
--- a/src/rabbit_exchange_type_direct.erl
+++ b/src/rabbit_exchange_type_direct.erl
@@ -34,7 +34,7 @@
 
 -behaviour(rabbit_exchange_type).
 
--export([description/0, publish/2]).
+-export([description/0, route/2]).
 -export([validate/1, create/1, recover/2, delete/2,
          add_binding/2, remove_bindings/2, assert_args_equivalence/2]).
 -include("rabbit_exchange_type_spec.hrl").
@@ -50,10 +50,9 @@ description() ->
     [{name, <<"direct">>},
      {description, <<"AMQP direct exchange, as per the AMQP specification">>}].
 
-publish(#exchange{name = Name}, Delivery =
-        #delivery{message = #basic_message{routing_key = RoutingKey}}) ->
-    rabbit_router:deliver(rabbit_router:match_routing_key(Name, RoutingKey),
-                          Delivery).
+route(#exchange{name = Name},
+      #delivery{message = #basic_message{routing_key = RoutingKey}}) ->
+    rabbit_router:match_routing_key(Name, RoutingKey).
 
 validate(_X) -> ok.
 create(_X) -> ok.
diff --git a/src/rabbit_exchange_type_fanout.erl b/src/rabbit_exchange_type_fanout.erl
index 94798c78fe..77ca9686c2 100644
--- a/src/rabbit_exchange_type_fanout.erl
+++ b/src/rabbit_exchange_type_fanout.erl
@@ -34,7 +34,7 @@
 
 -behaviour(rabbit_exchange_type).
 
--export([description/0, publish/2]).
+-export([description/0, route/2]).
 -export([validate/1, create/1, recover/2, delete/2, add_binding/2,
          remove_bindings/2, assert_args_equivalence/2]).
 -include("rabbit_exchange_type_spec.hrl").
@@ -50,8 +50,8 @@ description() ->
     [{name, <<"fanout">>},
      {description, <<"AMQP fanout exchange, as per the AMQP specification">>}].
 
-publish(#exchange{name = Name}, Delivery) ->
-    rabbit_router:deliver(rabbit_router:match_routing_key(Name, '_'), Delivery).
+route(#exchange{name = Name}, _Delivery) ->
+    rabbit_router:match_routing_key(Name, '_').
 
 validate(_X) -> ok.
 create(_X) -> ok.
diff --git a/src/rabbit_exchange_type_headers.erl b/src/rabbit_exchange_type_headers.erl
index 0a59a175cd..ec9e7ba468 100644
--- a/src/rabbit_exchange_type_headers.erl
+++ b/src/rabbit_exchange_type_headers.erl
@@ -35,7 +35,7 @@
 
 -behaviour(rabbit_exchange_type).
 
--export([description/0, publish/2]).
+-export([description/0, route/2]).
 -export([validate/1, create/1, recover/2, delete/2, add_binding/2,
          remove_bindings/2, assert_args_equivalence/2]).
 -include("rabbit_exchange_type_spec.hrl").
@@ -56,17 +56,14 @@ description() ->
     [{name, <<"headers">>},
      {description, <<"AMQP headers exchange, as per the AMQP specification">>}].
 
-publish(#exchange{name = Name},
-        Delivery = #delivery{message = #basic_message{content = Content}}) ->
+route(#exchange{name = Name},
+      #delivery{message = #basic_message{content = Content}}) ->
     Headers = case (Content#content.properties)#'P_basic'.headers of
                   undefined -> [];
                   H         -> rabbit_misc:sort_field_table(H)
               end,
-    rabbit_router:deliver(rabbit_router:match_bindings(
-                            Name, fun (#binding{args = Spec}) ->
-                                          headers_match(Spec, Headers)
-                                  end),
-                          Delivery).
+    rabbit_router:match_bindings(
+      Name, fun (#binding{args = Spec}) -> headers_match(Spec, Headers) end).
 
 default_headers_match_kind() -> all.
 
@@ -79,7 +76,7 @@ parse_x_match(Other) ->
 
 %% Horrendous matching algorithm. Depends for its merge-like
 %% (linear-time) behaviour on the lists:keysort
-%% (rabbit_misc:sort_field_table) that publish/1 and
+%% (rabbit_misc:sort_field_table) that route/1 and
 %% rabbit_binding:{add,remove}/2 do.
 %%
 %%                 !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
diff --git a/src/rabbit_exchange_type_topic.erl b/src/rabbit_exchange_type_topic.erl
index e796acf327..d3ecdd4dd2 100644
--- a/src/rabbit_exchange_type_topic.erl
+++ b/src/rabbit_exchange_type_topic.erl
@@ -34,7 +34,7 @@
 
 -behaviour(rabbit_exchange_type).
 
--export([description/0, publish/2]).
+-export([description/0, route/2]).
 -export([validate/1, create/1, recover/2, delete/2, add_binding/2,
          remove_bindings/2, assert_args_equivalence/2]).
 -include("rabbit_exchange_type_spec.hrl").
@@ -58,13 +58,12 @@ description() ->
     [{name, <<"topic">>},
      {description, <<"AMQP topic exchange, as per the AMQP specification">>}].
 
-publish(#exchange{name = Name}, Delivery =
+route(#exchange{name = Name},
         #delivery{message = #basic_message{routing_key = RoutingKey}}) ->
-    rabbit_router:deliver(rabbit_router:match_bindings(
-                            Name, fun (#binding{key = BindingKey}) ->
-                                          topic_matches(BindingKey, RoutingKey)
-                                  end),
-                          Delivery).
+    rabbit_router:match_bindings(Name,
+                                 fun (#binding{key = BindingKey}) ->
+                                         topic_matches(BindingKey, RoutingKey)
+                                 end).
 
 split_topic_key(Key) ->
     string:tokens(binary_to_list(Key), ".").
diff --git a/src/rabbit_limiter.erl b/src/rabbit_limiter.erl
index c323d7cef0..be1dcad1a6 100644
--- a/src/rabbit_limiter.erl
+++ b/src/rabbit_limiter.erl
@@ -37,7 +37,7 @@
          handle_info/2, prioritise_call/3]).
 -export([start_link/2]).
 -export([limit/2, can_send/3, ack/2, register/2, unregister/2]).
--export([get_limit/1, block/1, unblock/1]).
+-export([get_limit/1, block/1, unblock/1, is_blocked/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -55,6 +55,7 @@
 -spec(get_limit/1 :: (maybe_pid()) -> non_neg_integer()).
 -spec(block/1 :: (maybe_pid()) -> 'ok').
 -spec(unblock/1 :: (maybe_pid()) -> 'ok' | 'stopped').
+-spec(is_blocked/1 :: (maybe_pid()) -> boolean()).
 
 -endif.
 
@@ -119,6 +120,11 @@ unblock(undefined) ->
 unblock(LimiterPid) ->
     gen_server2:call(LimiterPid, unblock, infinity).
 
+is_blocked(undefined) ->
+    false;
+is_blocked(LimiterPid) ->
+    gen_server2:call(LimiterPid, is_blocked, infinity).
+
 %%----------------------------------------------------------------------------
 %% gen_server callbacks
 %%----------------------------------------------------------------------------
@@ -157,7 +163,10 @@ handle_call(unblock, _From, State) ->
     case maybe_notify(State, State#lim{blocked = false}) of
         {cont, State1} -> {reply, ok, State1};
         {stop, State1} -> {stop, normal, stopped, State1}
-    end.
+    end;
+
+handle_call(is_blocked, _From, State) ->
+    {reply, blocked(State), State}.
 
 handle_cast({ack, Count}, State = #lim{volume = Volume}) ->
     NewVolume = if Volume == 0 -> 0;
@@ -186,8 +195,8 @@ code_change(_, State, _) ->
 %%----------------------------------------------------------------------------
 
 maybe_notify(OldState, NewState) ->
-    case (limit_reached(OldState) orelse is_blocked(OldState)) andalso
-        not (limit_reached(NewState) orelse is_blocked(NewState)) of
+    case (limit_reached(OldState) orelse blocked(OldState)) andalso
+        not (limit_reached(NewState) orelse blocked(NewState)) of
         true  -> NewState1 = notify_queues(NewState),
                  {case NewState1#lim.prefetch_count of
                       0 -> stop;
@@ -199,7 +208,7 @@ maybe_notify(OldState, NewState) ->
 limit_reached(#lim{prefetch_count = Limit, volume = Volume}) ->
     Limit =/= 0 andalso Volume >= Limit.
 
-is_blocked(#lim{blocked = Blocked}) -> Blocked.
+blocked(#lim{blocked = Blocked}) -> Blocked.
 
 remember_queue(QPid, State = #lim{queues = Queues}) ->
     case dict:is_key(QPid, Queues) of
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index a321488897..577d206d1b 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -44,6 +44,9 @@
 
 -include("rabbit.hrl").
 
+-define(SCHEMA_VERSION_SET, []).
+-define(SCHEMA_VERSION_FILENAME, "schema_version").
+
 %%----------------------------------------------------------------------------
 
 -ifdef(use_specs).
@@ -91,6 +94,9 @@ init() ->
     ok = ensure_mnesia_running(),
     ok = ensure_mnesia_dir(),
     ok = init_db(read_cluster_nodes_config(), true),
+    ok = rabbit_misc:write_term_file(filename:join(
+                                       dir(), ?SCHEMA_VERSION_FILENAME),
+                                     [?SCHEMA_VERSION_SET]),
     ok.
 
 is_db_empty() ->
@@ -212,13 +218,15 @@ table_definitions() ->
        {match, #amqqueue{name = queue_name_match(), _='_'}}]}].
 
 binding_match() ->
-    #binding{queue_name = queue_name_match(),
-             exchange_name = exchange_name_match(),
+    #binding{source = exchange_name_match(),
+             destination = binding_destination_match(),
              _='_'}.
 reverse_binding_match() ->
-    #reverse_binding{queue_name = queue_name_match(),
-                     exchange_name = exchange_name_match(),
+    #reverse_binding{destination = binding_destination_match(),
+                     source = exchange_name_match(),
                      _='_'}.
+binding_destination_match() ->
+    resource_match('_').
 exchange_name_match() ->
     resource_match(exchange).
 queue_name_match() ->
@@ -241,7 +249,8 @@ ensure_mnesia_dir() ->
     case filelib:ensure_dir(MnesiaDir) of
         {error, Reason} ->
             throw({error, {cannot_create_mnesia_dir, MnesiaDir, Reason}});
-        ok -> ok
+        ok ->
+            ok
     end.
 
 ensure_mnesia_running() ->
diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index bbecbfe211..66cc06cf94 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -33,9 +33,9 @@
 
 -behaviour(gen_server2).
 
--export([start_link/4, write/4, read/3, contains/2, remove/2, release/2,
-         sync/3, client_init/2, client_terminate/2,
-         client_delete_and_terminate/3, successfully_recovered_state/1]).
+-export([start_link/4, successfully_recovered_state/1,
+         client_init/2, client_terminate/2, client_delete_and_terminate/3,
+         write/4, read/3, contains/2, remove/2, release/2, sync/3]).
 
 -export([sync/1, gc_done/4, set_maximum_since_use/2, gc/3]). %% internal
 
@@ -123,6 +123,11 @@
 -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/2 :: (server(), binary()) -> client_msstate()).
+-spec(client_terminate/2 :: (client_msstate(), server()) -> 'ok').
+-spec(client_delete_and_terminate/3 ::
+        (client_msstate(), server(), binary()) -> 'ok').
 -spec(write/4 :: (server(), rabbit_guid:guid(), msg(), client_msstate()) ->
                       rabbit_types:ok(client_msstate())).
 -spec(read/3 :: (server(), rabbit_guid:guid(), client_msstate()) ->
@@ -131,15 +136,11 @@
 -spec(remove/2 :: (server(), [rabbit_guid:guid()]) -> 'ok').
 -spec(release/2 :: (server(), [rabbit_guid:guid()]) -> 'ok').
 -spec(sync/3 :: (server(), [rabbit_guid:guid()], fun (() -> any())) -> 'ok').
+
+-spec(sync/1 :: (server()) -> 'ok').
 -spec(gc_done/4 :: (server(), non_neg_integer(), file_num(), file_num()) ->
                         'ok').
 -spec(set_maximum_since_use/2 :: (server(), non_neg_integer()) -> 'ok').
--spec(client_init/2 :: (server(), binary()) -> client_msstate()).
--spec(client_terminate/2 :: (client_msstate(), server()) -> 'ok').
--spec(client_delete_and_terminate/3 ::
-        (client_msstate(), server(), binary()) -> 'ok').
--spec(successfully_recovered_state/1 :: (server()) -> boolean()).
-
 -spec(gc/3 :: (non_neg_integer(), non_neg_integer(),
                {ets:tid(), file:filename(), atom(), any()}) ->
                    'concurrent_readers' | non_neg_integer()).
@@ -308,6 +309,31 @@ start_link(Server, Dir, ClientRefs, StartupFunState) ->
                            [Server, Dir, ClientRefs, StartupFunState],
                            [{timeout, infinity}]).
 
+successfully_recovered_state(Server) ->
+    gen_server2:call(Server, successfully_recovered_state, infinity).
+
+client_init(Server, Ref) ->
+    {IState, IModule, Dir, GCPid,
+     FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts} =
+        gen_server2:call(Server, {new_client_state, Ref}, infinity),
+    #client_msstate { file_handle_cache  = dict:new(),
+                      index_state        = IState,
+                      index_module       = IModule,
+                      dir                = Dir,
+                      gc_pid             = GCPid,
+                      file_handles_ets   = FileHandlesEts,
+                      file_summary_ets   = FileSummaryEts,
+                      dedup_cache_ets    = DedupCacheEts,
+                      cur_file_cache_ets = CurFileCacheEts }.
+
+client_terminate(CState, Server) ->
+    close_all_handles(CState),
+    ok = gen_server2:call(Server, client_terminate, infinity).
+
+client_delete_and_terminate(CState, Server, Ref) ->
+    close_all_handles(CState),
+    ok = gen_server2:cast(Server, {client_delete, Ref}).
+
 write(Server, Guid, Msg,
       CState = #client_msstate { cur_file_cache_ets = CurFileCacheEts }) ->
     ok = update_msg_cache(CurFileCacheEts, Guid, Msg),
@@ -325,7 +351,7 @@ read(Server, Guid,
                     Defer = fun() -> {gen_server2:call(
                                         Server, {read, Guid}, infinity),
                                       CState} end,
-                    case index_lookup(Guid, CState) of
+                    case index_lookup_positive_ref_count(Guid, CState) of
                         not_found   -> Defer();
                         MsgLocation -> client_read1(Server, MsgLocation, Defer,
                                                     CState)
@@ -345,7 +371,9 @@ remove(Server, Guids)  -> gen_server2:cast(Server, {remove, Guids}).
 release(_Server, [])   -> ok;
 release(Server, Guids) -> gen_server2:cast(Server, {release, Guids}).
 sync(Server, Guids, K) -> gen_server2:cast(Server, {sync, Guids, K}).
-sync(Server)           -> gen_server2:cast(Server, sync). %% internal
+
+sync(Server) ->
+    gen_server2:cast(Server, sync).
 
 gc_done(Server, Reclaimed, Source, Destination) ->
     gen_server2:cast(Server, {gc_done, Reclaimed, Source, Destination}).
@@ -353,31 +381,6 @@ gc_done(Server, Reclaimed, Source, Destination) ->
 set_maximum_since_use(Server, Age) ->
     gen_server2:cast(Server, {set_maximum_since_use, Age}).
 
-client_init(Server, Ref) ->
-    {IState, IModule, Dir, GCPid,
-     FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts} =
-        gen_server2:call(Server, {new_client_state, Ref}, infinity),
-    #client_msstate { file_handle_cache  = dict:new(),
-                      index_state        = IState,
-                      index_module       = IModule,
-                      dir                = Dir,
-                      gc_pid             = GCPid,
-                      file_handles_ets   = FileHandlesEts,
-                      file_summary_ets   = FileSummaryEts,
-                      dedup_cache_ets    = DedupCacheEts,
-                      cur_file_cache_ets = CurFileCacheEts }.
-
-client_terminate(CState, Server) ->
-    close_all_handles(CState),
-    ok = gen_server2:call(Server, client_terminate, infinity).
-
-client_delete_and_terminate(CState, Server, Ref) ->
-    close_all_handles(CState),
-    ok = gen_server2:cast(Server, {client_delete, Ref}).
-
-successfully_recovered_state(Server) ->
-    gen_server2:call(Server, successfully_recovered_state, infinity).
-
 %%----------------------------------------------------------------------------
 %% Client-side-only helpers
 %%----------------------------------------------------------------------------
@@ -577,8 +580,8 @@ init([Server, BaseDir, ClientRefs, StartupFunState]) ->
 
 prioritise_call(Msg, _From, _State) ->
     case Msg of
-        {new_client_state, _Ref}     -> 7;
         successfully_recovered_state -> 7;
+        {new_client_state, _Ref}     -> 7;
         {read, _Guid}                -> 2;
         _                            -> 0
     end.
@@ -591,13 +594,8 @@ prioritise_cast(Msg, _State) ->
         _                                            -> 0
     end.
 
-handle_call({read, Guid}, From, State) ->
-    State1 = read_message(Guid, From, State),
-    noreply(State1);
-
-handle_call({contains, Guid}, From, State) ->
-    State1 = contains_message(Guid, From, State),
-    noreply(State1);
+handle_call(successfully_recovered_state, _From, State) ->
+    reply(State #msstate.successfully_recovered, State);
 
 handle_call({new_client_state, CRef}, _From,
             State = #msstate { dir                = Dir,
@@ -613,52 +611,48 @@ handle_call({new_client_state, CRef}, _From,
            FileHandlesEts, FileSummaryEts, DedupCacheEts, CurFileCacheEts},
           State #msstate { client_refs = sets:add_element(CRef, ClientRefs) });
 
-handle_call(successfully_recovered_state, _From, State) ->
-    reply(State #msstate.successfully_recovered, State);
-
 handle_call(client_terminate, _From, State) ->
-    reply(ok, State).
+    reply(ok, State);
+
+handle_call({read, Guid}, From, State) ->
+    State1 = read_message(Guid, From, State),
+    noreply(State1);
+
+handle_call({contains, Guid}, From, State) ->
+    State1 = contains_message(Guid, From, State),
+    noreply(State1).
+
+handle_cast({client_delete, CRef},
+            State = #msstate { client_refs = ClientRefs }) ->
+    noreply(
+      State #msstate { client_refs = sets:del_element(CRef, ClientRefs) });
 
 handle_cast({write, Guid},
-            State = #msstate { current_file_handle = CurHdl,
-                               current_file        = CurFile,
-                               sum_valid_data      = SumValid,
-                               sum_file_size       = SumFileSize,
-                               file_summary_ets    = FileSummaryEts,
-                               cur_file_cache_ets  = CurFileCacheEts }) ->
+            State = #msstate { sum_valid_data     = SumValid,
+                               file_summary_ets   = FileSummaryEts,
+                               cur_file_cache_ets = CurFileCacheEts }) ->
     true = 0 =< ets:update_counter(CurFileCacheEts, Guid, {3, -1}),
     [{Guid, Msg, _CacheRefCount}] = ets:lookup(CurFileCacheEts, Guid),
     case index_lookup(Guid, State) of
         not_found ->
-            %% New message, lots to do
-            {ok, CurOffset} = file_handle_cache:current_virtual_offset(CurHdl),
-            {ok, TotalSize} = rabbit_msg_file:append(CurHdl, Guid, Msg),
-            ok = index_insert(#msg_location {
-                                guid = Guid, ref_count = 1, file = CurFile,
-                                offset = CurOffset, total_size = TotalSize },
-                              State),
-            [#file_summary { valid_total_size = ValidTotalSize,
-                             right            = undefined,
-                             locked           = false,
-                             file_size        = FileSize }] =
-                ets:lookup(FileSummaryEts, CurFile),
-            ValidTotalSize1 = ValidTotalSize + TotalSize,
-            true = ets:update_element(
-                     FileSummaryEts, CurFile,
-                     [{#file_summary.valid_total_size, ValidTotalSize1},
-                      {#file_summary.file_size,        FileSize + TotalSize}]),
-            NextOffset = CurOffset + TotalSize,
-            noreply(
-              maybe_roll_to_new_file(
-                NextOffset, State #msstate {
-                              sum_valid_data = SumValid + TotalSize,
-                              sum_file_size  = SumFileSize + TotalSize }));
+            write_message(Guid, Msg, State);
+        #msg_location { ref_count = 0, file = File, total_size = TotalSize } ->
+            case ets:lookup(FileSummaryEts, File) of
+                [#file_summary { locked = true }] ->
+                    ok = index_delete(Guid, State),
+                    write_message(Guid, Msg, State);
+                [#file_summary {}] ->
+                    ok = index_update_ref_count(Guid, 1, State),
+                    [_] = ets:update_counter(
+                            FileSummaryEts, File,
+                            [{#file_summary.valid_total_size, TotalSize}]),
+                    noreply(State #msstate {
+                              sum_valid_data = SumValid + TotalSize })
+            end;
         #msg_location { ref_count = RefCount } ->
             %% We already know about it, just update counter. Only
             %% update field otherwise bad interaction with concurrent GC
-            ok = index_update_fields(Guid,
-                                     {#msg_location.ref_count, RefCount + 1},
-                                     State),
+            ok = index_update_ref_count(Guid, RefCount + 1, State),
             noreply(State)
     end;
 
@@ -726,12 +720,7 @@ handle_cast({gc_done, Reclaimed, Src, Dst},
 
 handle_cast({set_maximum_since_use, Age}, State) ->
     ok = file_handle_cache:set_maximum_since_use(Age),
-    noreply(State);
-
-handle_cast({client_delete, CRef},
-            State = #msstate { client_refs = ClientRefs }) ->
-    noreply(
-      State #msstate { client_refs = sets:del_element(CRef, ClientRefs) }).
+    noreply(State).
 
 handle_info(timeout, State) ->
     noreply(internal_sync(State));
@@ -812,9 +801,31 @@ internal_sync(State = #msstate { current_file_handle = CurHdl,
               State1 #msstate { on_sync = [] }
     end.
 
+write_message(Guid, Msg,
+              State = #msstate { current_file_handle = CurHdl,
+                                 current_file        = CurFile,
+                                 sum_valid_data      = SumValid,
+                                 sum_file_size       = SumFileSize,
+                                 file_summary_ets    = FileSummaryEts }) ->
+    {ok, CurOffset} = file_handle_cache:current_virtual_offset(CurHdl),
+    {ok, TotalSize} = rabbit_msg_file:append(CurHdl, Guid, Msg),
+    ok = index_insert(
+           #msg_location { guid = Guid, ref_count = 1, file = CurFile,
+                           offset = CurOffset, total_size = TotalSize }, State),
+    [#file_summary { right = undefined, locked = false }] =
+        ets:lookup(FileSummaryEts, CurFile),
+    [_,_] = ets:update_counter(FileSummaryEts, CurFile,
+                               [{#file_summary.valid_total_size, TotalSize},
+                                {#file_summary.file_size,        TotalSize}]),
+    NextOffset = CurOffset + TotalSize,
+    noreply(maybe_roll_to_new_file(
+              NextOffset, State #msstate {
+                            sum_valid_data = SumValid    + TotalSize,
+                            sum_file_size  = SumFileSize + TotalSize })).
+
 read_message(Guid, From,
              State = #msstate { dedup_cache_ets = DedupCacheEts }) ->
-    case index_lookup(Guid, State) of
+    case index_lookup_positive_ref_count(Guid, State) of
         not_found ->
             gen_server2:reply(From, not_found),
             State;
@@ -887,7 +898,7 @@ read_from_disk(#msg_location { guid = Guid, ref_count = RefCount,
     {Msg, State1}.
 
 contains_message(Guid, From, State = #msstate { gc_active = GCActive }) ->
-    case index_lookup(Guid, State) of
+    case index_lookup_positive_ref_count(Guid, State) of
         not_found ->
             gen_server2:reply(From, false),
             State;
@@ -906,36 +917,30 @@ remove_message(Guid, State = #msstate { sum_valid_data   = SumValid,
                                         file_summary_ets = FileSummaryEts,
                                         dedup_cache_ets  = DedupCacheEts }) ->
     #msg_location { ref_count = RefCount, file = File,
-                    total_size = TotalSize } = index_lookup(Guid, State),
+                    total_size = TotalSize } =
+        index_lookup_positive_ref_count(Guid, State),
+    %% only update field, otherwise bad interaction with concurrent GC
+    Dec = fun () -> index_update_ref_count(Guid, RefCount - 1, State) end,
     case RefCount of
-        1 ->
-            %% don't remove from CUR_FILE_CACHE_ETS_NAME here because
-            %% there may be further writes in the mailbox for the same
-            %% msg.
-            ok = remove_cache_entry(DedupCacheEts, Guid),
-            [#file_summary { valid_total_size = ValidTotalSize,
-                             locked           = Locked }] =
-                ets:lookup(FileSummaryEts, File),
-            case Locked of
-                true ->
-                    add_to_pending_gc_completion({remove, Guid}, State);
-                false ->
-                    ok = index_delete(Guid, State),
-                    ValidTotalSize1 = ValidTotalSize - TotalSize,
-                    true =
-                        ets:update_element(
-                          FileSummaryEts, File,
-                          [{#file_summary.valid_total_size, ValidTotalSize1}]),
-                    State1 = delete_file_if_empty(File, State),
-                    State1 #msstate { sum_valid_data = SumValid - TotalSize }
-            end;
-        _ when 1 < RefCount ->
-            ok = decrement_cache(DedupCacheEts, Guid),
-            %% only update field, otherwise bad interaction with concurrent GC
-            ok = index_update_fields(Guid,
-                                     {#msg_location.ref_count, RefCount - 1},
-                                     State),
-            State
+        %% don't remove from CUR_FILE_CACHE_ETS_NAME here because
+        %% there may be further writes in the mailbox for the same
+        %% msg.
+        1 -> ok = remove_cache_entry(DedupCacheEts, Guid),
+             case ets:lookup(FileSummaryEts, File) of
+                 [#file_summary { locked = true } ] ->
+                     add_to_pending_gc_completion({remove, Guid}, State);
+                 [#file_summary {}] ->
+                     ok = Dec(),
+                     [_] = ets:update_counter(
+                             FileSummaryEts, File,
+                             [{#file_summary.valid_total_size, -TotalSize}]),
+                     delete_file_if_empty(
+                       File, State #msstate {
+                               sum_valid_data = SumValid - TotalSize })
+             end;
+        _ -> ok = decrement_cache(DedupCacheEts, Guid),
+             ok = Dec(),
+             State
     end.
 
 add_to_pending_gc_completion(
@@ -1106,6 +1111,16 @@ decrement_cache(DedupCacheEts, Guid) ->
 %% index
 %%----------------------------------------------------------------------------
 
+index_lookup_positive_ref_count(Key, State) ->
+    case index_lookup(Key, State) of
+        not_found                       -> not_found;
+        #msg_location { ref_count = 0 } -> not_found;
+        #msg_location {} = MsgLocation  -> MsgLocation
+    end.
+
+index_update_ref_count(Key, RefCount, State) ->
+    index_update_fields(Key, {#msg_location.ref_count, RefCount}, State).
+
 index_lookup(Key, #client_msstate { index_module = Index,
                                     index_state  = State }) ->
     Index:lookup(Key, State);
@@ -1498,6 +1513,10 @@ delete_file_if_empty(File, State = #msstate {
              end,
              true = mark_handle_to_close(FileHandlesEts, File),
              true = ets:delete(FileSummaryEts, File),
+             {ok, Messages, FileSize} =
+                 scan_file_for_valid_messages(Dir, filenum_to_name(File)),
+             [index_delete(Guid, State) ||
+                 {Guid, _TotalSize, _Offset} <- Messages],
              State1 = close_handle(File, State),
              ok = file:delete(form_filename(Dir, filenum_to_name(File))),
              State1 #msstate { sum_file_size = SumFileSize - FileSize };
@@ -1553,7 +1572,7 @@ combine_files(#file_summary { file             = Source,
     %%   copy back in, and then copy over from Source
     %% otherwise we just truncate straight away and copy over from Source
     {DestinationWorkList, DestinationValid} =
-        find_unremoved_messages_in_file(Destination, State),
+        load_and_vacuum_message_file(Destination, State),
     {DestinationContiguousTop, DestinationWorkListTail} =
         drop_contiguous_block_prefix(DestinationWorkList),
     case DestinationWorkListTail of
@@ -1579,8 +1598,7 @@ combine_files(#file_summary { file             = Source,
               ok = file_handle_cache:sync(DestinationHdl),
               ok = file_handle_cache:delete(TmpHdl)
     end,
-    {SourceWorkList, SourceValid} =
-        find_unremoved_messages_in_file(Source, State),
+    {SourceWorkList, SourceValid} = load_and_vacuum_message_file(Source, State),
     ok = copy_messages(SourceWorkList, DestinationValid, ExpectedSize,
                        SourceHdl, DestinationHdl, Destination, State),
     %% tidy up
@@ -1588,21 +1606,25 @@ combine_files(#file_summary { file             = Source,
     ok = file_handle_cache:delete(SourceHdl),
     ExpectedSize.
 
-find_unremoved_messages_in_file(File,
-                                {_FileSummaryEts, Dir, Index, IndexState}) ->
+load_and_vacuum_message_file(File, {_FileSummaryEts, Dir, Index, IndexState}) ->
     %% Messages here will be end-of-file at start-of-list
     {ok, Messages, _FileSize} =
         scan_file_for_valid_messages(Dir, filenum_to_name(File)),
     %% foldl will reverse so will end up with msgs in ascending offset order
-    lists:foldl(fun ({Guid, TotalSize, Offset}, Acc = {List, Size}) ->
-                        case Index:lookup(Guid, IndexState) of
-                            #msg_location { file = File, total_size = TotalSize,
-                                            offset = Offset } = Entry ->
-                                {[ Entry | List ], TotalSize + Size};
-                            _ ->
-                                Acc
-                        end
-                end, {[], 0}, Messages).
+    lists:foldl(
+      fun ({Guid, TotalSize, Offset}, Acc = {List, Size}) ->
+              case Index:lookup(Guid, IndexState) of
+                  #msg_location { file = File, total_size = TotalSize,
+                                  offset = Offset, ref_count = 0 } = Entry ->
+                      ok = Index:delete_object(Entry, IndexState),
+                      Acc;
+                  #msg_location { file = File, total_size = TotalSize,
+                                  offset = Offset } = Entry ->
+                      {[ Entry | List ], TotalSize + Size};
+                  _ ->
+                      Acc
+              end
+      end, {[], 0}, Messages).
 
 copy_messages(WorkList, InitOffset, FinalOffset, SourceHdl, DestinationHdl,
               Destination, {_FileSummaryEts, _Dir, Index, IndexState}) ->
diff --git a/src/rabbit_msg_store_ets_index.erl b/src/rabbit_msg_store_ets_index.erl
index 1eb3c11fb5..96be674c0b 100644
--- a/src/rabbit_msg_store_ets_index.erl
+++ b/src/rabbit_msg_store_ets_index.erl
@@ -35,7 +35,7 @@
 
 -export([new/1, recover/1,
          lookup/2, insert/2, update/2, update_fields/3, delete/2,
-         delete_by_file/2, terminate/1]).
+         delete_object/2, delete_by_file/2, terminate/1]).
 
 -define(MSG_LOC_NAME, rabbit_msg_store_ets_index).
 -define(FILENAME, "msg_store_index.ets").
@@ -79,6 +79,10 @@ delete(Key, State) ->
     true = ets:delete(State #state.table, Key),
     ok.
 
+delete_object(Obj, State) ->
+    true = ets:delete_object(State #state.table, Obj),
+    ok.
+
 delete_by_file(File, State) ->
     MatchHead = #msg_location { file = File, _ = '_' },
     ets:select_delete(State #state.table, [{MatchHead, [], [true]}]),
diff --git a/src/rabbit_multi.erl b/src/rabbit_multi.erl
index 5cfd6a5ca1..b48d0aa394 100644
--- a/src/rabbit_multi.erl
+++ b/src/rabbit_multi.erl
@@ -64,20 +64,20 @@ start() ->
                     io:format("done.~n"),
                     halt();
                 {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} ->
-                    error("invalid command '~s'",
-                          [lists:flatten(
-                             rabbit_misc:intersperse(" ", FullCommand))]),
+                    print_error("invalid command '~s'",
+                                [lists:flatten(
+                                   rabbit_misc:intersperse(" ", FullCommand))]),
                     usage();
                 timeout ->
-                    error("timeout starting some nodes.", []),
+                    print_error("timeout starting some nodes.", []),
                     halt(1);
                 Other ->
-                    error("~p", [Other]),
+                    print_error("~p", [Other]),
                     halt(2)
             end
     end.
 
-error(Format, Args) ->
+print_error(Format, Args) ->
     rabbit_misc:format_stderr("Error: " ++ Format ++ "~n", Args).
 
 parse_args([Command | Args]) ->
diff --git a/src/rabbit_net.erl b/src/rabbit_net.erl
index 2286896b7b..53d0d5cbf3 100644
--- a/src/rabbit_net.erl
+++ b/src/rabbit_net.erl
@@ -33,7 +33,7 @@
 -include("rabbit.hrl").
 
 -export([async_recv/3, close/1, controlling_process/2,
-        getstat/2, peername/1, port_command/2,
+        getstat/2, peername/1, peercert/1, port_command/2,
         send/2, sockname/1]).
 
 %%---------------------------------------------------------------------------
@@ -45,28 +45,29 @@
 -type(stat_option() ::
 	'recv_cnt' | 'recv_max' | 'recv_avg' | 'recv_oct' | 'recv_dvi' |
 	'send_cnt' | 'send_max' | 'send_avg' | 'send_oct' | 'send_pend').
--type(error() :: rabbit_types:error(any())).
+-type(ok_val_or_error(A) :: rabbit_types:ok_or_error2(A, any())).
+-type(ok_or_any_error() :: rabbit_types:ok_or_error(any())).
 -type(socket() :: port() | #ssl_socket{}).
 
 -spec(async_recv/3 ::
         (socket(), integer(), timeout()) -> rabbit_types:ok(any())).
--spec(close/1 :: (socket()) -> rabbit_types:ok_or_error(any())).
--spec(controlling_process/2 ::
-        (socket(), pid()) -> rabbit_types:ok_or_error(any())).
+-spec(close/1 :: (socket()) -> ok_or_any_error()).
+-spec(controlling_process/2 :: (socket(), pid()) -> ok_or_any_error()).
 -spec(port_command/2 :: (socket(), iolist()) -> 'true').
 -spec(send/2 ::
-        (socket(), binary() | iolist()) -> rabbit_types:ok_or_error(any())).
+        (socket(), binary() | iolist()) -> ok_or_any_error()).
 -spec(peername/1 ::
         (socket())
-        -> rabbit_types:ok({inet:ip_address(), rabbit_networking:ip_port()}) |
-           error()).
+        -> ok_val_or_error({inet:ip_address(), rabbit_networking:ip_port()})).
+-spec(peercert/1 ::
+        (socket())
+        -> 'nossl' | ok_val_or_error(rabbit_ssl:certificate())).
 -spec(sockname/1 ::
         (socket())
-        -> rabbit_types:ok({inet:ip_address(), rabbit_networking:ip_port()}) |
-           error()).
+        -> ok_val_or_error({inet:ip_address(), rabbit_networking:ip_port()})).
 -spec(getstat/2 ::
         (socket(), [stat_option()])
-        -> rabbit_types:ok([{stat_option(), integer()}]) | error()).
+        -> ok_val_or_error([{stat_option(), integer()}])).
 
 -endif.
 
@@ -108,6 +109,11 @@ peername(Sock) when ?IS_SSL(Sock) ->
 peername(Sock) when is_port(Sock) ->
     inet:peername(Sock).
 
+peercert(Sock) when ?IS_SSL(Sock) ->
+    ssl:peercert(Sock#ssl_socket.ssl);
+peercert(Sock) when is_port(Sock) ->
+    nossl.
+
 port_command(Sock, Data) when ?IS_SSL(Sock) ->
     case ssl:send(Sock#ssl_socket.ssl, Data) of
         ok              -> self() ! {inet_reply, Sock, ok},
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 6dbd54d2bc..db5c71f62a 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -46,8 +46,6 @@
 
 -include("rabbit.hrl").
 -include_lib("kernel/include/inet.hrl").
--include_lib("ssl/src/ssl_record.hrl").
-
 
 -define(RABBIT_TCP_OPTS, [
         binary,
@@ -120,26 +118,7 @@ boot_ssl() ->
                                                  end}
                                    | SslOptsConfig]
                 end,
-            % In R13B04 and R14A (at least), rc4 is incorrectly implemented.
-            CipherSuites = proplists:get_value(ciphers,
-                                               SslOpts,
-                                               ssl:cipher_suites()),
-            FilteredCipherSuites =
-                [C || C <- CipherSuites,
-                      begin
-                          SuiteCode =
-                              if is_tuple(C) -> ssl_cipher:suite(C);
-                                 is_list(C)  -> ssl_cipher:openssl_suite(C)
-                              end,
-                          SP = ssl_cipher:security_parameters(
-                              SuiteCode,
-                              #security_parameters{}),
-                          SP#security_parameters.bulk_cipher_algorithm =/= ?RC4
-                      end],
-            SslOpts1 = [{ciphers, FilteredCipherSuites}
-                        | [{K, V} || {K, V} <- SslOpts, K =/= ciphers]],
-            [start_ssl_listener(Host, Port, SslOpts1)
-                || {Host, Port} <- SslListeners],
+            [start_ssl_listener(Host, Port, SslOpts) || {Host, Port} <- SslListeners],
             ok
     end.
 
diff --git a/src/rabbit_plugin_activator.erl b/src/rabbit_plugin_activator.erl
index b23776cd74..88300ab448 100644
--- a/src/rabbit_plugin_activator.erl
+++ b/src/rabbit_plugin_activator.erl
@@ -72,7 +72,8 @@ start() ->
     %% applications along the way
     AllApps = case catch sets:to_list(expand_dependencies(RequiredApps)) of
                   {failed_to_load_app, App, Err} ->
-                      error("failed to load application ~s:~n~p", [App, Err]);
+                      terminate("failed to load application ~s:~n~p",
+                                [App, Err]);
                   AppList ->
                       AppList
               end,
@@ -90,7 +91,7 @@ start() ->
 
     %% Compile the script
     ScriptFile = RootName ++ ".script",
-    case systools:make_script(RootName, [local, silent]) of
+    case systools:make_script(RootName, [local, silent, exref]) of
         {ok, Module, Warnings} ->
             %% This gets lots of spurious no-source warnings when we
             %% have .ez files, so we want to supress them to prevent
@@ -116,19 +117,20 @@ start() ->
             end,
             ok;
         {error, Module, Error} ->
-            error("generation of boot script file ~s failed:~n~s",
-                  [ScriptFile, Module:format_error(Error)])
+            terminate("generation of boot script file ~s failed:~n~s",
+                      [ScriptFile, Module:format_error(Error)])
     end,
 
     case post_process_script(ScriptFile) of
         ok -> ok;
         {error, Reason} ->
-            error("post processing of boot script file ~s failed:~n~w",
-                  [ScriptFile, Reason])
+            terminate("post processing of boot script file ~s failed:~n~w",
+                      [ScriptFile, Reason])
     end,
     case systools:script2boot(RootName) of
         ok    -> ok;
-        error -> error("failed to compile boot script file ~s", [ScriptFile])
+        error -> terminate("failed to compile boot script file ~s",
+                           [ScriptFile])
     end,
     io:format("~w plugins activated:~n", [length(PluginApps)]),
     [io:format("* ~s-~s~n", [App, proplists:get_value(App, AppVersions)])
@@ -190,11 +192,11 @@ unpack_ez_plugins(SrcDir, DestDir) ->
     %% Eliminate the contents of the destination directory
     case delete_recursively(DestDir) of
         ok         -> ok;
-        {error, E} -> error("Could not delete dir ~s (~p)", [DestDir, E])
+        {error, E} -> terminate("Could not delete dir ~s (~p)", [DestDir, E])
     end,
     case filelib:ensure_dir(DestDir ++ "/") of
         ok          -> ok;
-        {error, E2} -> error("Could not create dir ~s (~p)", [DestDir, E2])
+        {error, E2} -> terminate("Could not create dir ~s (~p)", [DestDir, E2])
     end,
     [unpack_ez_plugin(PluginName, DestDir) ||
         PluginName <- filelib:wildcard(SrcDir ++ "/*.ez")].
@@ -261,6 +263,6 @@ process_entry(Entry = {apply,{application,start_boot,[rabbit,permanent]}}) ->
 process_entry(Entry) ->
     [Entry].
 
-error(Fmt, Args) ->
+terminate(Fmt, Args) ->
     io:format("ERROR: " ++ Fmt ++ "~n", Args),
     halt(1).
diff --git a/src/rabbit_queue_collector.erl b/src/rabbit_queue_collector.erl
index 0a49b94d09..6ac402c8c4 100644
--- a/src/rabbit_queue_collector.erl
+++ b/src/rabbit_queue_collector.erl
@@ -38,7 +38,7 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--record(state, {queues}).
+-record(state, {queues, delete_from}).
 
 -include("rabbit.hrl").
 
@@ -66,32 +66,39 @@ delete_all(CollectorPid) ->
 %%----------------------------------------------------------------------------
 
 init([]) ->
-    {ok, #state{queues = dict:new()}}.
+    {ok, #state{queues = dict:new(), delete_from = undefined}}.
 
 %%--------------------------------------------------------------------------
 
 handle_call({register, Q}, _From,
-            State = #state{queues = Queues}) ->
+            State = #state{queues = Queues, delete_from = Deleting}) ->
     MonitorRef = erlang:monitor(process, Q#amqqueue.pid),
-    {reply, ok,
-     State#state{queues = dict:store(MonitorRef, Q, Queues)}};
-
-handle_call(delete_all, _From, State = #state{queues = Queues}) ->
-    [rabbit_misc:with_exit_handler(
-       fun () -> ok end,
-       fun () ->
-               erlang:demonitor(MonitorRef),
-               rabbit_amqqueue:delete(Q, false, false)
-       end)
-     || {MonitorRef, Q} <- dict:to_list(Queues)],
-    {reply, ok, State}.
+    case Deleting of
+        undefined -> ok;
+        _         -> rabbit_amqqueue:delete_immediately(Q)
+    end,
+    {reply, ok, State#state{queues = dict:store(MonitorRef, Q, Queues)}};
+
+handle_call(delete_all, From, State = #state{queues      = Queues,
+                                             delete_from = undefined}) ->
+    case dict:size(Queues) of
+        0 -> {reply, ok, State#state{delete_from = From}};
+        _ -> [rabbit_amqqueue:delete_immediately(Q)
+              || {_MRef, Q} <- dict:to_list(Queues)],
+             {noreply, State#state{delete_from = From}}
+    end.
 
 handle_cast(Msg, State) ->
     {stop, {unhandled_cast, Msg}, State}.
 
 handle_info({'DOWN', MonitorRef, process, _DownPid, _Reason},
-            State = #state{queues = Queues}) ->
-    {noreply, State#state{queues = dict:erase(MonitorRef, Queues)}}.
+            State = #state{queues = Queues, delete_from = Deleting}) ->
+    Queues1 = dict:erase(MonitorRef, Queues),
+    case Deleting =/= undefined andalso dict:size(Queues1) =:= 0 of
+        true  -> gen_server:reply(Deleting, ok);
+        false -> ok
+    end,
+    {noreply, State#state{queues = Queues1}}.
 
 terminate(_Reason, _State) ->
     ok.
diff --git a/src/rabbit_queue_index.erl b/src/rabbit_queue_index.erl
index d6b8bb2889..0b98290ccd 100644
--- a/src/rabbit_queue_index.erl
+++ b/src/rabbit_queue_index.erl
@@ -159,7 +159,9 @@
 
 -define(PUB, {_, _}). %% {Guid, IsPersistent}
 
--define(READ_MODE, [binary, raw, read, {read_ahead, ?SEGMENT_TOTAL_SIZE}]).
+-define(READ_MODE, [binary, raw, read]).
+-define(READ_AHEAD_MODE, [{read_ahead, ?SEGMENT_TOTAL_SIZE} | ?READ_MODE]).
+-define(WRITE_MODE, [write | ?READ_MODE]).
 
 %%----------------------------------------------------------------------------
 
@@ -220,8 +222,13 @@
 %% public API
 %%----------------------------------------------------------------------------
 
-init(Name, Recover, MsgStoreRecovered, ContainsCheckFun) ->
-    State = #qistate { dir = Dir } = blank_state(Name, not Recover),
+init(Name, false, _MsgStoreRecovered, _ContainsCheckFun) ->
+    State = #qistate { dir = Dir } = blank_state(Name),
+    false = filelib:is_file(Dir), %% is_file == is file or dir
+    {0, [], State};
+
+init(Name, true, MsgStoreRecovered, ContainsCheckFun) ->
+    State = #qistate { dir = Dir } = blank_state(Name),
     Terms = case read_shutdown_terms(Dir) of
                 {error, _}   -> [];
                 {ok, Terms1} -> Terms1
@@ -356,15 +363,8 @@ recover(DurableQueues) ->
 %% startup and shutdown
 %%----------------------------------------------------------------------------
 
-blank_state(QueueName, EnsureFresh) ->
-    StrName = queue_name_to_dir_name(QueueName),
-    Dir = filename:join(queues_dir(), StrName),
-    ok = case EnsureFresh of
-             true  -> false = filelib:is_file(Dir), %% is_file == is file or dir
-                      ok;
-             false -> ok
-         end,
-    ok = filelib:ensure_dir(filename:join(Dir, "nothing")),
+blank_state(QueueName) ->
+    Dir = filename:join(queues_dir(), queue_name_to_dir_name(QueueName)),
     {ok, MaxJournal} =
         application:get_env(rabbit, queue_index_max_journal_entries),
     #qistate { dir                 = Dir,
@@ -373,17 +373,21 @@ blank_state(QueueName, EnsureFresh) ->
                dirty_count         = 0,
                max_journal_entries = MaxJournal }.
 
+clean_file_name(Dir) -> filename:join(Dir, ?CLEAN_FILENAME).
+
 detect_clean_shutdown(Dir) ->
-    case file:delete(filename:join(Dir, ?CLEAN_FILENAME)) of
+    case file:delete(clean_file_name(Dir)) of
         ok              -> true;
         {error, enoent} -> false
     end.
 
 read_shutdown_terms(Dir) ->
-    rabbit_misc:read_term_file(filename:join(Dir, ?CLEAN_FILENAME)).
+    rabbit_misc:read_term_file(clean_file_name(Dir)).
 
 store_clean_shutdown(Terms, Dir) ->
-    rabbit_misc:write_term_file(filename:join(Dir, ?CLEAN_FILENAME), Terms).
+    CleanFileName = clean_file_name(Dir),
+    ok = filelib:ensure_dir(CleanFileName),
+    rabbit_misc:write_term_file(CleanFileName, Terms).
 
 init_clean(RecoveredCounts, State) ->
     %% Load the journal. Since this is a clean recovery this (almost)
@@ -500,7 +504,7 @@ queue_index_walker({next, Gatherer}) when is_pid(Gatherer) ->
 
 queue_index_walker_reader(QueueName, Gatherer) ->
     State = #qistate { segments = Segments, dir = Dir } =
-        recover_journal(blank_state(QueueName, false)),
+        recover_journal(blank_state(QueueName)),
     [ok = segment_entries_foldr(
             fun (_RelSeq, {{Guid, true}, _IsDelivered, no_ack}, ok) ->
                     gatherer:in(Gatherer, {Guid, 1});
@@ -578,7 +582,7 @@ append_journal_to_segment(#segment { journal_entries = JEntries,
                                      path = Path } = Segment) ->
     case array:sparse_size(JEntries) of
         0 -> Segment;
-        _ -> {ok, Hdl} = file_handle_cache:open(Path, [write | ?READ_MODE],
+        _ -> {ok, Hdl} = file_handle_cache:open(Path, ?WRITE_MODE,
                                                 [{write_buffer, infinity}]),
              array:sparse_foldl(fun write_entry_to_segment/3, Hdl, JEntries),
              ok = file_handle_cache:close(Hdl),
@@ -588,7 +592,8 @@ append_journal_to_segment(#segment { journal_entries = JEntries,
 get_journal_handle(State = #qistate { journal_handle = undefined,
                                       dir = Dir }) ->
     Path = filename:join(Dir, ?JOURNAL_FILENAME),
-    {ok, Hdl} = file_handle_cache:open(Path, [write | ?READ_MODE],
+    ok = filelib:ensure_dir(Path),
+    {ok, Hdl} = file_handle_cache:open(Path, ?WRITE_MODE,
                                        [{write_buffer, infinity}]),
     {Hdl, State #qistate { journal_handle = Hdl }};
 get_journal_handle(State = #qistate { journal_handle = Hdl }) ->
@@ -785,7 +790,7 @@ segment_entries_foldr(Fun, Init,
 load_segment(KeepAcked, #segment { path = Path }) ->
     case filelib:is_file(Path) of
         false -> {array_new(), 0};
-        true  -> {ok, Hdl} = file_handle_cache:open(Path, ?READ_MODE, []),
+        true  -> {ok, Hdl} = file_handle_cache:open(Path, ?READ_AHEAD_MODE, []),
                  {ok, 0} = file_handle_cache:position(Hdl, bof),
                  Res = load_segment_entries(KeepAcked, Hdl, array_new(), 0),
                  ok = file_handle_cache:close(Hdl),
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 745e008349..29004bd5fb 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -66,6 +66,8 @@
                           send_pend, state, channels]).
 
 -define(CREATION_EVENT_KEYS, [pid, address, port, peer_address, peer_port,
+                              peer_cert_subject, peer_cert_issuer,
+                              peer_cert_validity,
                               protocol, user, vhost, timeout, frame_max,
                               client_properties]).
 
@@ -813,27 +815,26 @@ infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
 i(pid, #v1{}) ->
     self();
 i(address, #v1{sock = Sock}) ->
-    {ok, {A, _}} = rabbit_net:sockname(Sock),
-    A;
+    socket_info(fun rabbit_net:sockname/1, fun ({A, _}) -> A end, Sock);
 i(port, #v1{sock = Sock}) ->
-    {ok, {_, P}} = rabbit_net:sockname(Sock),
-    P;
+    socket_info(fun rabbit_net:sockname/1, fun ({_, P}) -> P end, Sock);
 i(peer_address, #v1{sock = Sock}) ->
-    {ok, {A, _}} = rabbit_net:peername(Sock),
-    A;
+    socket_info(fun rabbit_net:peername/1, fun ({A, _}) -> A end, Sock);
 i(peer_port, #v1{sock = Sock}) ->
-    {ok, {_, P}} = rabbit_net:peername(Sock),
-    P;
+    socket_info(fun rabbit_net:peername/1, fun ({_, P}) -> P end, Sock);
+i(peer_cert_issuer, #v1{sock = Sock}) ->
+    cert_info(fun rabbit_ssl:peer_cert_issuer/1, Sock);
+i(peer_cert_subject, #v1{sock = Sock}) ->
+    cert_info(fun rabbit_ssl:peer_cert_subject/1, Sock);
+i(peer_cert_validity, #v1{sock = Sock}) ->
+    cert_info(fun rabbit_ssl:peer_cert_validity/1, Sock);
 i(SockStat, #v1{sock = Sock}) when SockStat =:= recv_oct;
                                    SockStat =:= recv_cnt;
                                    SockStat =:= send_oct;
                                    SockStat =:= send_cnt;
                                    SockStat =:= send_pend ->
-    case rabbit_net:getstat(Sock, [SockStat]) of
-        {ok, [{SockStat, StatVal}]} -> StatVal;
-        {error, einval}             -> undefined;
-        {error, Error}              -> throw({cannot_get_socket_stats, Error})
-    end;
+    socket_info(fun () -> rabbit_net:getstat(Sock, [SockStat]) end,
+                fun ([{_, I}]) -> I end);
 i(state, #v1{connection_state = S}) ->
     S;
 i(channels, #v1{}) ->
@@ -858,6 +859,22 @@ i(client_properties, #v1{connection = #connection{
 i(Item, #v1{}) ->
     throw({bad_argument, Item}).
 
+socket_info(Get, Select, Sock) ->
+    socket_info(fun() -> Get(Sock) end, Select).
+
+socket_info(Get, Select) ->
+    case Get() of
+        {ok,    T} -> Select(T);
+        {error, _} -> ''
+    end.
+
+cert_info(F, Sock) ->
+    case rabbit_net:peercert(Sock) of
+        nossl                -> '';
+        {error, no_peercert} -> '';
+        {ok, Cert}           -> F(Cert)
+    end.
+
 %%--------------------------------------------------------------------------
 
 send_to_new_channel(Channel, AnalyzedFrame, State) ->
@@ -891,7 +908,7 @@ handle_exception(State = #v1{connection_state = CS}, Channel, Reason) ->
 send_exception(State = #v1{connection = #connection{protocol = Protocol}},
                Channel, Reason) ->
     {ShouldClose, CloseChannel, CloseMethod} =
-        map_exception(Channel, Reason, Protocol),
+        rabbit_binary_generator:map_exception(Channel, Reason, Protocol),
     NewState = case ShouldClose of
                    true  -> terminate_channels(),
                             close_connection(State);
@@ -901,47 +918,6 @@ send_exception(State = #v1{connection = #connection{protocol = Protocol}},
            NewState#v1.sock, CloseChannel, CloseMethod, Protocol),
     NewState.
 
-map_exception(Channel, Reason, Protocol) ->
-    {SuggestedClose, ReplyCode, ReplyText, FailedMethod} =
-        lookup_amqp_exception(Reason, Protocol),
-    ShouldClose = SuggestedClose or (Channel == 0),
-    {ClassId, MethodId} = case FailedMethod of
-                              {_, _} -> FailedMethod;
-                              none   -> {0, 0};
-                              _      -> Protocol:method_id(FailedMethod)
-                          end,
-    {CloseChannel, CloseMethod} =
-        case ShouldClose of
-            true -> {0, #'connection.close'{reply_code = ReplyCode,
-                                            reply_text = ReplyText,
-                                            class_id = ClassId,
-                                            method_id = MethodId}};
-            false -> {Channel, #'channel.close'{reply_code = ReplyCode,
-                                                reply_text = ReplyText,
-                                                class_id = ClassId,
-                                                method_id = MethodId}}
-        end,
-    {ShouldClose, CloseChannel, CloseMethod}.
-
-lookup_amqp_exception(#amqp_error{name        = Name,
-                                  explanation = Expl,
-                                  method      = Method},
-                      Protocol) ->
-    {ShouldClose, Code, Text} = Protocol:lookup_amqp_exception(Name),
-    ExplBin = amqp_exception_explanation(Text, Expl),
-    {ShouldClose, Code, ExplBin, Method};
-lookup_amqp_exception(Other, Protocol) ->
-    rabbit_log:warning("Non-AMQP exit reason '~p'~n", [Other]),
-    {ShouldClose, Code, Text} = Protocol:lookup_amqp_exception(internal_error),
-    {ShouldClose, Code, Text, none}.
-
-amqp_exception_explanation(Text, Expl) ->
-    ExplBin = list_to_binary(Expl),
-    CompleteTextBin = <<Text/binary, " - ", ExplBin/binary>>,
-    if size(CompleteTextBin) > 255 -> <<CompleteTextBin:252/binary, "...">>;
-       true                        -> CompleteTextBin
-    end.
-
 internal_emit_stats(State = #v1{stats_timer = StatsTimer}) ->
     rabbit_event:notify(connection_stats, infos(?STATISTICS_KEYS, State)),
     State#v1{stats_timer = rabbit_event:reset_stats_timer(StatsTimer)}.
diff --git a/src/rabbit_router.erl b/src/rabbit_router.erl
index bd57f73726..00df1ce1f7 100644
--- a/src/rabbit_router.erl
+++ b/src/rabbit_router.erl
@@ -39,26 +39,27 @@
 
 -ifdef(use_specs).
 
--export_type([routing_key/0, routing_result/0]).
+-export_type([routing_key/0, routing_result/0, match_result/0]).
 
 -type(routing_key() :: binary()).
 -type(routing_result() :: 'routed' | 'unroutable' | 'not_delivered').
 -type(qpids() :: [pid()]).
+-type(match_result() :: [rabbit_types:binding_destination()]).
 
--spec(deliver/2 ::
-        (qpids(), rabbit_types:delivery()) -> {routing_result(), qpids()}).
--spec(match_bindings/2 :: (rabbit_exchange:name(),
+-spec(deliver/2 :: ([rabbit_amqqueue:name()], rabbit_types:delivery()) ->
+                        {routing_result(), qpids()}).
+-spec(match_bindings/2 :: (rabbit_types:binding_source(),
                            fun ((rabbit_types:binding()) -> boolean())) ->
-    qpids()).
--spec(match_routing_key/2 :: (rabbit_exchange:name(), routing_key() | '_') ->
-                                  qpids()).
+    match_result()).
+-spec(match_routing_key/2 :: (rabbit_types:binding_source(),
+                              routing_key() | '_') -> match_result()).
 
 -endif.
 
 %%----------------------------------------------------------------------------
 
-deliver(QPids, Delivery = #delivery{mandatory = false,
-                                    immediate = false}) ->
+deliver(QNames, Delivery = #delivery{mandatory = false,
+                                     immediate = false}) ->
     %% optimisation: when Mandatory = false and Immediate = false,
     %% rabbit_amqqueue:deliver will deliver the message to the queue
     %% process asynchronously, and return true, which means all the
@@ -66,11 +67,13 @@ deliver(QPids, Delivery = #delivery{mandatory = false,
     %% fire-and-forget cast here and return the QPids - the semantics
     %% is preserved. This scales much better than the non-immediate
     %% case below.
+    QPids = lookup_qpids(QNames),
     delegate:invoke_no_result(
       QPids, fun (Pid) -> rabbit_amqqueue:deliver(Pid, Delivery) end),
     {routed, QPids};
 
-deliver(QPids, Delivery) ->
+deliver(QNames, Delivery) ->
+    QPids = lookup_qpids(QNames),
     {Success, _} =
         delegate:invoke(QPids,
                         fun (Pid) ->
@@ -82,31 +85,23 @@ deliver(QPids, Delivery) ->
                    {Routed, Handled}).
 
 %% TODO: Maybe this should be handled by a cursor instead.
-%% TODO: This causes a full scan for each entry with the same exchange
-match_bindings(Name, Match) ->
-    Query = qlc:q([QName || #route{binding = Binding = #binding{
-                                               exchange_name = XName,
-                                               queue_name = QName}} <-
-                                mnesia:table(rabbit_route),
-                            XName == Name,
-                            Match(Binding)]),
-    lookup_qpids(mnesia:async_dirty(fun qlc:e/1, [Query])).
-
-match_routing_key(Name, RoutingKey) ->
-    MatchHead = #route{binding = #binding{exchange_name = Name,
-                                          queue_name = '$1',
-                                          key = RoutingKey,
-                                          _ = '_'}},
-    lookup_qpids(mnesia:dirty_select(rabbit_route, [{MatchHead, [], ['$1']}])).
-
-lookup_qpids(Queues) ->
-    lists:foldl(
-      fun (Key, Acc) ->
-              case mnesia:dirty_read({rabbit_queue, Key}) of
-                  [#amqqueue{pid = QPid}] -> [QPid | Acc];
-                  []                      -> Acc
-              end
-      end, [], lists:usort(Queues)).
+%% TODO: This causes a full scan for each entry with the same source
+match_bindings(SrcName, Match) ->
+    Query = qlc:q([DestinationName ||
+                      #route{binding = Binding = #binding{
+                                         source      = SrcName1,
+                                         destination = DestinationName}} <-
+                          mnesia:table(rabbit_route),
+                      SrcName == SrcName1,
+                      Match(Binding)]),
+    mnesia:async_dirty(fun qlc:e/1, [Query]).
+
+match_routing_key(SrcName, RoutingKey) ->
+    MatchHead = #route{binding = #binding{source      = SrcName,
+                                          destination = '$1',
+                                          key         = RoutingKey,
+                                          _           = '_'}},
+    mnesia:dirty_select(rabbit_route, [{MatchHead, [], ['$1']}]).
 
 %%--------------------------------------------------------------------
 
@@ -117,3 +112,11 @@ fold_deliveries({_,  false},{_, Handled}) -> {true, Handled}.
 check_delivery(true, _   , {false, []}) -> {unroutable, []};
 check_delivery(_   , true, {_    , []}) -> {not_delivered, []};
 check_delivery(_   , _   , {_    , Qs}) -> {routed, Qs}.
+
+lookup_qpids(QNames) ->
+    lists:foldl(fun (QName, QPids) ->
+                        case mnesia:dirty_read({rabbit_queue, QName}) of
+                            [#amqqueue{pid = QPid}] -> [QPid | QPids];
+                            []                      -> QPids
+                        end
+                end, [], QNames).
diff --git a/src/rabbit_ssl.erl b/src/rabbit_ssl.erl
new file mode 100644
index 0000000000..be451af631
--- /dev/null
+++ b/src/rabbit_ssl.erl
@@ -0,0 +1,173 @@
+%%   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_ssl).
+
+-include("rabbit.hrl").
+
+-include_lib("public_key/include/public_key.hrl").
+-include_lib("ssl/src/ssl_int.hrl").
+
+-export([peer_cert_issuer/1, peer_cert_subject/1, peer_cert_validity/1]).
+
+%%--------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-export_type([certificate/0]).
+
+-type(certificate() :: binary()).
+
+-spec(peer_cert_issuer/1   :: (certificate()) -> string()).
+-spec(peer_cert_subject/1  :: (certificate()) -> string()).
+-spec(peer_cert_validity/1 :: (certificate()) -> string()).
+
+-endif.
+
+%%--------------------------------------------------------------------------
+%% High-level functions used by reader
+%%--------------------------------------------------------------------------
+
+%% Return a string describing the certificate's issuer.
+peer_cert_issuer(Cert) ->
+    cert_info(fun(#'OTPCertificate' {
+                     tbsCertificate = #'OTPTBSCertificate' {
+                       issuer = Issuer }}) ->
+                      format_rdn_sequence(Issuer)
+              end, Cert).
+
+%% Return a string describing the certificate's subject, as per RFC4514.
+peer_cert_subject(Cert) ->
+    cert_info(fun(#'OTPCertificate' {
+                     tbsCertificate = #'OTPTBSCertificate' {
+                       subject = Subject }}) ->
+                      format_rdn_sequence(Subject)
+              end, Cert).
+
+%% Return a string describing the certificate's validity.
+peer_cert_validity(Cert) ->
+    cert_info(fun(#'OTPCertificate' {
+                     tbsCertificate = #'OTPTBSCertificate' {
+                       validity = {'Validity', Start, End} }}) ->
+                      lists:flatten(
+                        io_lib:format("~s - ~s", [format_asn1_value(Start),
+                                                  format_asn1_value(End)]))
+              end, Cert).
+
+%%--------------------------------------------------------------------------
+
+cert_info(F, Cert) ->
+    F(case public_key:pkix_decode_cert(Cert, otp) of
+          {ok, DecCert} -> DecCert;
+          DecCert       -> DecCert
+      end).
+
+%%--------------------------------------------------------------------------
+%% Formatting functions
+%%--------------------------------------------------------------------------
+
+%% Format and rdnSequence as a RFC4514 subject string.
+format_rdn_sequence({rdnSequence, Seq}) ->
+    lists:flatten(
+      rabbit_misc:intersperse(
+        ",", lists:reverse([format_complex_rdn(RDN) || RDN <- Seq]))).
+
+%% Format an RDN set.
+format_complex_rdn(RDNs) ->
+    lists:flatten(
+      rabbit_misc:intersperse("+", [format_rdn(RDN) || RDN <- RDNs])).
+
+%% Format an RDN.  If the type name is unknown, use the dotted decimal
+%% representation.  See RFC4514, section 2.3.
+format_rdn(#'AttributeTypeAndValue'{type = T, value = V}) ->
+    FV = escape_rdn_value(format_asn1_value(V)),
+    Fmts = [{?'id-at-surname'                , "SN"},
+            {?'id-at-givenName'              , "GIVENNAME"},
+            {?'id-at-initials'               , "INITIALS"},
+            {?'id-at-generationQualifier'    , "GENERATIONQUALIFIER"},
+            {?'id-at-commonName'             , "CN"},
+            {?'id-at-localityName'           , "L"},
+            {?'id-at-stateOrProvinceName'    , "ST"},
+            {?'id-at-organizationName'       , "O"},
+            {?'id-at-organizationalUnitName' , "OU"},
+            {?'id-at-title'                  , "TITLE"},
+            {?'id-at-countryName'            , "C"},
+            {?'id-at-serialNumber'           , "SERIALNUMBER"},
+            {?'id-at-pseudonym'              , "PSEUDONYM"},
+            {?'id-domainComponent'           , "DC"},
+            {?'id-emailAddress'              , "EMAILADDRESS"},
+            {?'street-address'               , "STREET"}],
+    case proplists:lookup(T, Fmts) of
+        {_, Fmt} ->
+            io_lib:format(Fmt ++ "=~s", [FV]);
+        none when is_tuple(T) ->
+            TypeL = [io_lib:format("~w", [X]) || X <- tuple_to_list(T)],
+            io_lib:format("~s:~s", [rabbit_misc:intersperse(".", TypeL), FV]);
+        none ->
+            io_lib:format("~p:~s", [T, FV])
+    end.
+
+%% Escape a string as per RFC4514.
+escape_rdn_value(V) ->
+    escape_rdn_value(V, start).
+
+escape_rdn_value([], _) ->
+    [];
+escape_rdn_value([C | S], start) when C =:= $ ; C =:= $# ->
+    [$\\, C | escape_rdn_value(S, middle)];
+escape_rdn_value(S, start) ->
+    escape_rdn_value(S, middle);
+escape_rdn_value([$ ], middle) ->
+    [$\\, $ ];
+escape_rdn_value([C | S], middle) when C =:= $"; C =:= $+; C =:= $,; C =:= $;;
+                                       C =:= $<; C =:= $>; C =:= $\\ ->
+    [$\\, C | escape_rdn_value(S, middle)];
+escape_rdn_value([C | S], middle) when C < 32 ; C =:= 127 ->
+    %% only U+0000 needs escaping, but for display purposes it's handy
+    %% to escape all non-printable chars
+    lists:flatten(io_lib:format("\\~2.16.0B", [C])) ++
+        escape_rdn_value(S, middle);
+escape_rdn_value([C | S], middle) ->
+    [C | escape_rdn_value(S, middle)].
+
+%% Get the string representation of an OTPCertificate field.
+format_asn1_value({ST, S}) when ST =:= teletexString; ST =:= printableString;
+                                ST =:= universalString; ST =:= utf8String;
+                                ST =:= bmpString ->
+    if is_binary(S) -> binary_to_list(S);
+       true         -> S
+    end;
+format_asn1_value({utcTime, [Y1, Y2, M1, M2, D1, D2, H1, H2,
+                            Min1, Min2, S1, S2, $Z]}) ->
+    io_lib:format("20~c~c-~c~c-~c~cT~c~c:~c~c:~c~cZ",
+                  [Y1, Y2, M1, M2, D1, D2, H1, H2, Min1, Min2, S1, S2]);
+format_asn1_value(V) ->
+    io_lib:format("~p", [V]).
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index a72656b73b..1b47cdb71c 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -75,7 +75,6 @@ all_tests() ->
     passed = maybe_run_cluster_dependent_tests(),
     passed.
 
-
 maybe_run_cluster_dependent_tests() ->
     SecondaryNode = rabbit_misc:makenode("hare"),
 
@@ -1041,11 +1040,11 @@ test_server_status() ->
     %% list bindings
     ok = info_action(list_bindings, rabbit_binding:info_keys(), true),
     %% misc binding listing APIs
-    [_|_] = rabbit_binding:list_for_exchange(
+    [_|_] = rabbit_binding:list_for_source(
               rabbit_misc:r(<<"/">>, exchange, <<"">>)),
-    [_] = rabbit_binding:list_for_queue(
+    [_] = rabbit_binding:list_for_destination(
               rabbit_misc:r(<<"/">>, queue, <<"foo">>)),
-    [_] = rabbit_binding:list_for_exchange_and_queue(
+    [_] = rabbit_binding:list_for_source_and_destination(
             rabbit_misc:r(<<"/">>, exchange, <<"">>),
             rabbit_misc:r(<<"/">>, queue, <<"foo">>)),
 
diff --git a/src/rabbit_types.erl b/src/rabbit_types.erl
index 0b6a15ec83..b971a63f72 100644
--- a/src/rabbit_types.erl
+++ b/src/rabbit_types.erl
@@ -39,9 +39,11 @@
               delivery/0, content/0, decoded_content/0, undecoded_content/0,
               unencoded_content/0, encoded_content/0, vhost/0, ctag/0,
               amqp_error/0, r/1, r2/2, r3/3, listener/0,
-              binding/0, amqqueue/0, exchange/0, connection/0, protocol/0,
-              user/0, ok/1, error/1, ok_or_error/1, ok_or_error2/2,
-              ok_pid_or_error/0, channel_exit/0, connection_exit/0]).
+              binding/0, binding_source/0, binding_destination/0,
+              amqqueue/0, exchange/0,
+              connection/0, protocol/0, user/0, ok/1, error/1, ok_or_error/1,
+              ok_or_error2/2, ok_pid_or_error/0, channel_exit/0,
+              connection_exit/0]).
 
 -type(channel_exit() :: no_return()).
 -type(connection_exit() :: no_return()).
@@ -113,11 +115,14 @@
                 host     :: rabbit_networking:hostname(),
                 port     :: rabbit_networking:ip_port()}).
 
+-type(binding_source() :: rabbit_exchange:name()).
+-type(binding_destination() :: rabbit_amqqueue:name() | rabbit_exchange:name()).
+
 -type(binding() ::
-      #binding{exchange_name    :: rabbit_exchange:name(),
-               queue_name       :: rabbit_amqqueue:name(),
-               key              :: rabbit_binding:key(),
-               args             :: rabbit_framing:amqp_table()}).
+      #binding{source      :: rabbit_exchange:name(),
+               destination :: binding_destination(),
+               key         :: rabbit_binding:key(),
+               args        :: rabbit_framing:amqp_table()}).
 
 -type(amqqueue() ::
       #amqqueue{name            :: rabbit_amqqueue:name(),
diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 30d3a8aec1..cbc71bcc5c 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -472,23 +472,30 @@ delete_and_terminate(State) ->
     a(State2 #vqstate { index_state       = IndexState1,
                         msg_store_clients = undefined }).
 
-purge(State = #vqstate { q4 = Q4, index_state = IndexState, len = Len }) ->
+purge(State = #vqstate { q4               = Q4,
+                         index_state      = IndexState,
+                         len              = Len,
+                         persistent_count = PCount }) ->
     %% TODO: when there are no pending acks, which is a common case,
     %% we could simply wipe the qi instead of issuing delivers and
     %% acks for all the messages.
-    IndexState1 = remove_queue_entries(fun rabbit_misc:queue_fold/3, Q4,
-                                       IndexState),
-    State1 = #vqstate { q1 = Q1, index_state = IndexState2 } =
-        purge_betas_and_deltas(State #vqstate { q4          = queue:new(),
+    {LensByStore, IndexState1} = remove_queue_entries(
+                                   fun rabbit_misc:queue_fold/3, Q4,
+                                   orddict:new(), IndexState),
+    {LensByStore1, State1 = #vqstate { q1 = Q1, index_state = IndexState2 }} =
+        purge_betas_and_deltas(LensByStore,
+                               State #vqstate { q4          = queue:new(),
                                                 index_state = IndexState1 }),
-    IndexState3 = remove_queue_entries(fun rabbit_misc:queue_fold/3, Q1,
-                                       IndexState2),
+    {LensByStore2, IndexState3} = remove_queue_entries(
+                                    fun rabbit_misc:queue_fold/3, Q1,
+                                    LensByStore1, IndexState2),
+    PCount1 = PCount - find_persistent_count(LensByStore2),
     {Len, a(State1 #vqstate { q1               = queue:new(),
                               index_state      = IndexState3,
                               len              = 0,
                               ram_msg_count    = 0,
                               ram_index_count  = 0,
-                              persistent_count = 0 })}.
+                              persistent_count = PCount1 })}.
 
 publish(Msg, State) ->
     {_SeqId, State1} = publish(Msg, false, false, State),
@@ -957,26 +964,30 @@ tx_commit_index(State = #vqstate { on_sync = #sync {
     reduce_memory_use(
       State1 #vqstate { index_state = IndexState1, on_sync = ?BLANK_SYNC }).
 
-purge_betas_and_deltas(State = #vqstate { q3          = Q3,
+purge_betas_and_deltas(LensByStore,
+                       State = #vqstate { q3          = Q3,
                                           index_state = IndexState }) ->
     case bpqueue:is_empty(Q3) of
-        true  -> State;
-        false -> IndexState1 = remove_queue_entries(fun beta_fold/3, Q3,
-                                                    IndexState),
-                 purge_betas_and_deltas(
-                   maybe_deltas_to_betas(
-                     State #vqstate { q3          = bpqueue:new(),
-                                      index_state = IndexState1 }))
+        true  -> {LensByStore, State};
+        false -> {LensByStore1, IndexState1} = remove_queue_entries(
+                                                 fun beta_fold/3, Q3,
+                                                 LensByStore, IndexState),
+                 purge_betas_and_deltas(LensByStore1,
+                                        maybe_deltas_to_betas(
+                                          State #vqstate {
+                                            q3          = bpqueue:new(),
+                                            index_state = IndexState1 }))
     end.
 
-remove_queue_entries(Fold, Q, IndexState) ->
+remove_queue_entries(Fold, Q, LensByStore, IndexState) ->
     {GuidsByStore, Delivers, Acks} =
         Fold(fun remove_queue_entries1/2, {orddict:new(), [], []}, Q),
     ok = orddict:fold(fun (MsgStore, Guids, ok) ->
                               rabbit_msg_store:remove(MsgStore, Guids)
                       end, ok, GuidsByStore),
-    rabbit_queue_index:ack(Acks,
-                           rabbit_queue_index:deliver(Delivers, IndexState)).
+    {sum_guids_by_store_to_len(LensByStore, GuidsByStore),
+     rabbit_queue_index:ack(Acks,
+                            rabbit_queue_index:deliver(Delivers, IndexState))}.
 
 remove_queue_entries1(
   #msg_status { guid = Guid, seq_id = SeqId,
@@ -991,6 +1002,12 @@ remove_queue_entries1(
      cons_if(IndexOnDisk andalso not IsDelivered, SeqId, Delivers),
      cons_if(IndexOnDisk, SeqId, Acks)}.
 
+sum_guids_by_store_to_len(LensByStore, GuidsByStore) ->
+    orddict:fold(
+      fun (MsgStore, Guids, LensByStore1) ->
+              orddict:update_counter(MsgStore, length(Guids), LensByStore1)
+      end, LensByStore, GuidsByStore).
+
 %%----------------------------------------------------------------------------
 %% Internal gubbins for publishing
 %%----------------------------------------------------------------------------
@@ -1117,10 +1134,8 @@ ack(MsgStoreFun, Fun, AckTags, State) ->
     ok = orddict:fold(fun (MsgStore, Guids, ok) ->
                               MsgStoreFun(MsgStore, Guids)
                       end, ok, GuidsByStore),
-    PCount1 = PCount - case orddict:find(?PERSISTENT_MSG_STORE, GuidsByStore) of
-                           error       -> 0;
-                           {ok, Guids} -> length(Guids)
-                       end,
+    PCount1 = PCount - find_persistent_count(sum_guids_by_store_to_len(
+                                               orddict:new(), GuidsByStore)),
     State1 #vqstate { index_state      = IndexState1,
                       persistent_count = PCount1 }.
 
@@ -1132,6 +1147,12 @@ accumulate_ack(SeqId, {IsPersistent, Guid}, {SeqIdsAcc, Dict}) ->
     {cons_if(IsPersistent, SeqId, SeqIdsAcc),
      rabbit_misc:orddict_cons(find_msg_store(IsPersistent), Guid, Dict)}.
 
+find_persistent_count(LensByStore) ->
+    case orddict:find(?PERSISTENT_MSG_STORE, LensByStore) of
+        error     -> 0;
+        {ok, Len} -> Len
+    end.
+
 %%----------------------------------------------------------------------------
 %% Phase changes
 %%----------------------------------------------------------------------------
diff --git a/src/vm_memory_monitor.erl b/src/vm_memory_monitor.erl
index e658f005a3..9eb9d0a6fd 100644
--- a/src/vm_memory_monitor.erl
+++ b/src/vm_memory_monitor.erl
@@ -47,7 +47,7 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--export([update/0, get_total_memory/0,
+-export([update/0, get_total_memory/0, get_vm_limit/0,
          get_check_interval/0, set_check_interval/1,
          get_vm_memory_high_watermark/0, set_vm_memory_high_watermark/1,
          get_memory_limit/0]).
@@ -76,7 +76,7 @@
 -spec(update/0 :: () -> 'ok').
 -spec(get_total_memory/0 :: () -> (non_neg_integer() | 'unknown')).
 -spec(get_vm_limit/0 :: () -> non_neg_integer()).
--spec(get_memory_limit/0 :: () -> (non_neg_integer() | 'undefined')).
+-spec(get_memory_limit/0 :: () -> non_neg_integer()).
 -spec(get_check_interval/0 :: () -> non_neg_integer()).
 -spec(set_check_interval/1 :: (non_neg_integer()) -> 'ok').
 -spec(get_vm_memory_high_watermark/0 :: () -> float()).
@@ -84,7 +84,6 @@
 
 -endif.
 
-
 %%----------------------------------------------------------------------------
 %% Public API
 %%----------------------------------------------------------------------------
@@ -296,6 +295,12 @@ get_total_memory({unix, sunos}) ->
     Dict = dict:from_list(lists:map(fun parse_line_sunos/1, Lines)),
     dict:fetch('Memory size', Dict);
 
+get_total_memory({unix, aix}) ->
+    File = cmd("/usr/bin/vmstat -v"),
+    Lines = string:tokens(File, "\n"),
+    Dict = dict:from_list(lists:map(fun parse_line_aix/1, Lines)),
+    dict:fetch('memory pages', Dict) * 4096;
+
 get_total_memory(_OsType) ->
     unknown.
 
@@ -341,6 +346,17 @@ parse_line_sunos(Line) ->
         [Name] -> {list_to_atom(Name), none}
     end.
 
+%% Lines look like " 12345 memory pages"
+%% or              "  80.1 maxpin percentage"
+parse_line_aix(Line) ->
+    [Value | NameWords] = string:tokens(Line, " "),
+    Name = string:join(NameWords, " "),
+    {list_to_atom(Name),
+     case lists:member($., Value) of
+         true  -> trunc(list_to_float(Value));
+         false -> list_to_integer(Value)
+     end}.
+
 freebsd_sysctl(Def) ->
     list_to_integer(cmd("/sbin/sysctl -n " ++ Def) -- "\n").