De-bitrot by merging default in. Passes rabbit_tests:all_tests() and

the ../rabbitmq-test test suite.  Some unused variables appear to have crept
in during previous commits to the branch; will fix in another commit.

13 files changed, 365 insertions, 329 deletions

diff --git a/Makefile b/Makefile
index 5d43ee8a..7ed4611c 100644
--- a/Makefile
+++ b/Makefile
@@ -52,7 +52,7 @@ TARGET_SRC_DIR=dist/$(TARBALL_NAME)
 
 SIBLING_CODEGEN_DIR=../rabbitmq-codegen/
 AMQP_CODEGEN_DIR=$(shell [ -d $(SIBLING_CODEGEN_DIR) ] && echo $(SIBLING_CODEGEN_DIR) || echo codegen)
-AMQP_SPEC_JSON_PATH=$(AMQP_CODEGEN_DIR)/amqp-0.8.json
+AMQP_SPEC_JSON_PATH=$(AMQP_CODEGEN_DIR)/amqp-0.9.1.json
 
 ERL_CALL=erl_call -sname $(RABBITMQ_NODENAME) -e
 
diff --git a/codegen.py b/codegen.py
index 96109610..b5c91cbd 100644
--- a/codegen.py
+++ b/codegen.py
@@ -368,6 +368,7 @@ def genHrl(spec):
     printFileHeader()
     print "-define(PROTOCOL_VERSION_MAJOR, %d)." % (spec.major)
     print "-define(PROTOCOL_VERSION_MINOR, %d)." % (spec.minor)
+    print "-define(PROTOCOL_VERSION_REVISION, %d)." % (spec.revision)
     print "-define(PROTOCOL_PORT, %d)." % (spec.port)
 
     for (c,v,cls) in spec.constants:
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index e2980eff..3baf8c1a 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -49,9 +49,9 @@
 
 -record(resource, {virtual_host, kind, name}).
 
--record(exchange, {name, type, durable, auto_delete, arguments}).
+-record(exchange, {name, type, durable, arguments}).
 
--record(amqqueue, {name, durable, auto_delete, arguments, pid}).
+-record(amqqueue, {name, durable, auto_delete, exclusive_owner = none, arguments, pid}).
 
 %% mnesia doesn't like unary records, so we add a dummy 'value' field
 -record(route, {binding, value = const}).
@@ -102,16 +102,16 @@
                   write     :: regexp(),
                   read      :: regexp()}).
 -type(amqqueue() ::
-      #amqqueue{name          :: queue_name(),
-                durable       :: boolean(),
-                auto_delete   :: boolean(),
-                arguments     :: amqp_table(),
-                pid           :: maybe(pid())}).
+      #amqqueue{name            :: queue_name(),
+                durable         :: boolean(),
+                auto_delete     :: boolean(),
+                exclusive_owner :: maybe(pid()),
+                arguments       :: amqp_table(),
+                pid             :: maybe(pid())}).
 -type(exchange() ::
       #exchange{name        :: exchange_name(),
                 type        :: exchange_type(),
                 durable     :: boolean(),
-                auto_delete :: boolean(),
                 arguments   :: amqp_table()}).
 -type(binding() ::
       #binding{exchange_name    :: exchange_name(),
diff --git a/src/rabbit.erl b/src/rabbit.erl
index 35d3ce4a..ac883a1b 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -372,9 +372,10 @@ print_banner() ->
               "| ~s  +---+   |~n"
               "|                   |~n"
               "+-------------------+~n"
-              "AMQP ~p-~p~n~s~n~s~n~n",
+              "AMQP ~p-~p-~p~n~s~n~s~n~n",
               [Product, string:right([$v|Version], ProductLen),
                ?PROTOCOL_VERSION_MAJOR, ?PROTOCOL_VERSION_MINOR,
+               ?PROTOCOL_VERSION_REVISION,
                ?COPYRIGHT_MESSAGE, ?INFORMATION_MESSAGE]),
     Settings = [{"node",           node()},
                 {"app descriptor", app_location()},
diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl
index a445f441..23b84afb 100644
--- a/src/rabbit_access_control.erl
+++ b/src/rabbit_access_control.erl
@@ -240,7 +240,7 @@ add_vhost(VHostPath) ->
                                             write),
                           [rabbit_exchange:declare(
                              rabbit_misc:r(VHostPath, exchange, Name),
-                             Type, true, false, []) ||
+                             Type, true, []) ||
                               {Name,Type} <-
                                   [{<<"">>,           direct},
                                    {<<"amq.direct">>, direct},
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 3f25d72e..2611d189 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -31,14 +31,13 @@
 
 -module(rabbit_amqqueue).
 
--export([start/0, recover/0, declare/4, delete/3, purge/1]).
+-export([start/0, recover/0, declare/5, delete/3, purge/1]).
 -export([internal_declare/2, internal_delete/1]).
 -export([pseudo_queue/2]).
 -export([lookup/1, with/2, with_or_die/2,
          stat/1, stat_all/0, deliver/2, redeliver/2, requeue/3, ack/4]).
 -export([list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2]).
 -export([consumers/1, consumers_all/1]).
--export([claim_queue/2]).
 -export([basic_get/3, basic_consume/8, basic_cancel/4]).
 -export([notify_sent/2, unblock/2]).
 -export([commit_all/2, rollback_all/2, notify_down_all/2, limit_all/3]).
@@ -64,7 +63,7 @@
 
 -spec(start/0 :: () -> 'ok').
 -spec(recover/0 :: () -> 'ok').
--spec(declare/4 :: (queue_name(), boolean(), boolean(), amqp_table()) ->
+-spec(declare/5 :: (queue_name(), boolean(), boolean(), amqp_table(), maybe(pid())) ->
              amqqueue()).
 -spec(lookup/1 :: (queue_name()) -> {'ok', amqqueue()} | not_found()).
 -spec(with/2 :: (queue_name(), qfun(A)) -> A | not_found()).
@@ -96,7 +95,6 @@
 -spec(rollback_all/2 :: ([pid()], txn()) -> ok_or_errors()).
 -spec(notify_down_all/2 :: ([pid()], pid()) -> ok_or_errors()).
 -spec(limit_all/3 :: ([pid()], pid(), pid() | 'undefined') -> ok_or_errors()).
--spec(claim_queue/2 :: (amqqueue(), pid()) -> 'ok' | 'locked').
 -spec(basic_get/3 :: (amqqueue(), pid(), boolean()) ->
              {'ok', non_neg_integer(), msg()} | 'empty').
 -spec(basic_consume/8 ::
@@ -128,24 +126,39 @@ recover() ->
     ok = recover_durable_queues(),
     ok.
 
+shared_or_live_owner(none) ->
+     true;
+shared_or_live_owner(Owner) when is_pid(Owner) ->
+    rpc:call(node(Owner), erlang, is_process_alive, [Owner]).
+
 recover_durable_queues() ->
     Node = node(),
     lists:foreach(
-      fun (RecoveredQ) ->
-              Q = start_queue_process(RecoveredQ),
+      fun (RecoveredQ = #amqqueue{ exclusive_owner = Owner }) ->
               %% We need to catch the case where a client connected to
               %% another node has deleted the queue (and possibly
               %% re-created it).
-              case rabbit_misc:execute_mnesia_transaction(
-                     fun () -> case mnesia:match_object(
-                                      rabbit_durable_queue, RecoveredQ, read) of
-                                   [_] -> ok = store_queue(Q),
-                                          true;
-                                   []  -> false
-                               end
-                     end) of
-                  true  -> ok;
-                  false -> exit(Q#amqqueue.pid, shutdown)
+              DoIfSameQueue =
+                  fun (Action) ->
+                          rabbit_misc:execute_mnesia_transaction(
+                            fun () -> case mnesia:match_object(
+                                             rabbit_durable_queue, RecoveredQ, read) of
+                                          [_] -> ok = Action(),
+                                                 true;
+                                          []  -> false
+                                      end
+                            end)
+                  end,
+              case shared_or_live_owner(Owner) of
+                  true ->
+                      Q = start_queue_process(RecoveredQ), 
+                      case DoIfSameQueue(fun () -> store_queue(Q) end) of
+                          true  -> ok;
+                          false -> exit(Q#amqqueue.pid, shutdown)
+                      end;
+                  false ->
+                      DoIfSameQueue(
+                        fun () -> internal_delete2(RecoveredQ#amqqueue.name) end)
               end
       end,
       %% TODO: use dirty ops instead
@@ -157,11 +170,12 @@ recover_durable_queues() ->
         end)),
     ok.
 
-declare(QueueName, Durable, AutoDelete, Args) ->
+declare(QueueName, Durable, AutoDelete, Args, Owner) ->
     Q = start_queue_process(#amqqueue{name = QueueName,
                                       durable = Durable,
                                       auto_delete = AutoDelete,
                                       arguments = Args,
+                                      exclusive_owner = Owner,
                                       pid = none}),
     internal_declare(Q, true).
 
@@ -206,7 +220,7 @@ start_queue_process(Q) ->
 add_default_binding(#amqqueue{name = QueueName}) ->
     Exchange = rabbit_misc:r(QueueName, exchange, <<>>),
     RoutingKey = QueueName#resource.name,
-    rabbit_exchange:add_binding(Exchange, QueueName, RoutingKey, []),
+    rabbit_exchange:add_binding(Exchange, QueueName, RoutingKey, [], fun (_X, _Q) -> ok end),
     ok.
 
 lookup(Name) ->
@@ -281,7 +295,7 @@ redeliver(QPid, Messages) ->
     gen_server2:cast(QPid, {redeliver, Messages}).
 
 requeue(QPid, MsgIds, ChPid) ->
-    gen_server2:cast(QPid, {requeue, MsgIds, ChPid}).
+    gen_server2:call(QPid, {requeue, MsgIds, ChPid}).
 
 ack(QPid, Txn, MsgIds, ChPid) ->
     gen_server2:pcast(QPid, 8, {ack, Txn, MsgIds, ChPid}).
@@ -312,9 +326,6 @@ limit_all(QPids, ChPid, LimiterPid) ->
       fun (QPid) -> gen_server2:cast(QPid, {limit, ChPid, LimiterPid}) end,
       QPids).
 
-claim_queue(#amqqueue{pid = QPid}, ReaderPid) ->
-    gen_server2:call(QPid, {claim_queue, ReaderPid}, infinity).
-
 basic_get(#amqqueue{pid = QPid}, ChPid, NoAck) ->
     gen_server2:call(QPid, {basic_get, ChPid, NoAck}, infinity).
 
@@ -334,16 +345,17 @@ notify_sent(QPid, ChPid) ->
 unblock(QPid, ChPid) ->
     gen_server2:pcast(QPid, 8, {unblock, ChPid}).
 
+internal_delete2(QueueName) ->
+    ok = rabbit_exchange:delete_queue_bindings(QueueName),
+    ok = mnesia:delete({rabbit_queue, QueueName}),
+    ok = mnesia:delete({rabbit_durable_queue, QueueName}).
+
 internal_delete(QueueName) ->
     rabbit_misc:execute_mnesia_transaction(
       fun () ->
               case mnesia:wread({rabbit_queue, QueueName}) of
                   []  -> {error, not_found};
-                  [_] ->
-                      ok = rabbit_exchange:delete_queue_bindings(QueueName),
-                      ok = mnesia:delete({rabbit_queue, QueueName}),
-                      ok = mnesia:delete({rabbit_durable_queue, QueueName}),
-                      ok
+                  [_] -> internal_delete2(QueueName)
               end
       end).
 
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index e4791f95..6c5f4757 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -49,7 +49,6 @@
 
 % Queue's state
 -record(q, {q,
-            owner,
             exclusive_consumer,
             has_had_consumers,
             next_msg_id,
@@ -99,8 +98,11 @@ info_keys() -> ?INFO_KEYS.
 
 init(Q) ->
     ?LOGDEBUG("Queue starting - ~p~n", [Q]),
+    case Q#amqqueue.exclusive_owner of
+        none      -> ok;
+        ReaderPid -> erlang:monitor(process, ReaderPid)
+    end,
     {ok, #q{q = Q,
-            owner = none,
             exclusive_consumer = none,
             has_had_consumers = false,
             next_msg_id = 1,
@@ -354,10 +356,6 @@ cancel_holder(ChPid, ConsumerTag, {ChPid, ConsumerTag}) ->
 cancel_holder(_ChPid, _ConsumerTag, Holder) ->
     Holder.
 
-check_queue_owner(none,           _)         -> ok;
-check_queue_owner({ReaderPid, _}, ReaderPid) -> ok;
-check_queue_owner({_,         _}, _)         -> mismatch.
-
 check_exclusive_access({_ChPid, _ConsumerTag}, _ExclusiveConsume, _State) ->
     in_use;
 check_exclusive_access(none, false, _State) ->
@@ -513,9 +511,9 @@ i(auto_delete, #q{q = #amqqueue{auto_delete = AutoDelete}}) -> AutoDelete;
 i(arguments,   #q{q = #amqqueue{arguments   = Arguments}})  -> Arguments;
 i(pid, _) ->
     self();
-i(owner_pid, #q{owner = none}) ->
+i(owner_pid, #q{q = #amqqueue{exclusive_owner = none}}) ->
     '';
-i(owner_pid, #q{owner = {ReaderPid, _MonitorRef}}) ->
+i(owner_pid, #q{q = #amqqueue{exclusive_owner = ReaderPid}}) ->
     ReaderPid;
 i(exclusive_consumer_pid, #q{exclusive_consumer = none}) ->
     '';
@@ -637,49 +635,43 @@ handle_call({basic_get, ChPid, NoAck}, _From,
 
 handle_call({basic_consume, NoAck, ReaderPid, ChPid, LimiterPid,
              ConsumerTag, ExclusiveConsume, OkMsg},
-            _From, State = #q{owner = Owner,
-                              exclusive_consumer = ExistingHolder}) ->
-    case check_queue_owner(Owner, ReaderPid) of
-        mismatch ->
-            reply({error, queue_owned_by_another_connection}, State);
+            _From, State = #q{exclusive_consumer = ExistingHolder}) ->
+    case check_exclusive_access(ExistingHolder, ExclusiveConsume,
+                                State) of
+        in_use ->
+            reply({error, exclusive_consume_unavailable}, State);
         ok ->
-            case check_exclusive_access(ExistingHolder, ExclusiveConsume,
-                                        State) of
-                in_use ->
-                    reply({error, exclusive_consume_unavailable}, State);
-                ok ->
-                    C = #cr{consumer_count = ConsumerCount} = ch_record(ChPid),
-                    Consumer = #consumer{tag = ConsumerTag,
-                                         ack_required = not(NoAck)},
-                    store_ch_record(C#cr{consumer_count = ConsumerCount +1,
-                                         limiter_pid = LimiterPid}),
-                    case ConsumerCount of
-                        0 -> ok = rabbit_limiter:register(LimiterPid, self());
-                        _ -> ok
-                    end,
-                    ExclusiveConsumer = case ExclusiveConsume of
-                                            true  -> {ChPid, ConsumerTag};
-                                            false -> ExistingHolder
-                                        end,
-                    State1 = State#q{has_had_consumers = true,
-                                     exclusive_consumer = ExclusiveConsumer},
-                    ok = maybe_send_reply(ChPid, OkMsg),
-                    State2 =
-                        case is_ch_blocked(C) of
-                            true  -> State1#q{
-                                       blocked_consumers =
-                                       add_consumer(
-                                         ChPid, Consumer,
-                                         State1#q.blocked_consumers)};
-                            false -> run_message_queue(
-                                       State1#q{
-                                         active_consumers =
-                                         add_consumer(
-                                           ChPid, Consumer,
-                                           State1#q.active_consumers)})
-                        end,
-                    reply(ok, State2)
-            end
+            C = #cr{consumer_count = ConsumerCount} = ch_record(ChPid),
+            Consumer = #consumer{tag = ConsumerTag,
+                                 ack_required = not(NoAck)},
+            store_ch_record(C#cr{consumer_count = ConsumerCount +1,
+                                 limiter_pid = LimiterPid}),
+            case ConsumerCount of
+                0 -> rabbit_limiter:register(LimiterPid, self());
+                _ -> ok
+            end,
+            ExclusiveConsumer = case ExclusiveConsume of
+                                    true  -> {ChPid, ConsumerTag};
+                                    false -> ExistingHolder
+                                end,
+            State1 = State#q{has_had_consumers = true,
+                             exclusive_consumer = ExclusiveConsumer},
+            ok = maybe_send_reply(ChPid, OkMsg),
+            State2 =
+                case is_ch_blocked(C) of
+                    true  -> State1#q{
+                               blocked_consumers =
+                               add_consumer(
+                                 ChPid, Consumer,
+                                 State1#q.blocked_consumers)};
+                    false -> run_message_queue(
+                               State1#q{
+                                 active_consumers =
+                                 add_consumer(
+                                   ChPid, Consumer,
+                                   State1#q.active_consumers)})
+                end,
+            reply(ok, State2)
     end;
 
 handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, _From,
@@ -735,29 +727,20 @@ handle_call(purge, _From, State = #q{message_buffer = MessageBuffer}) ->
     reply({ok, queue:len(MessageBuffer)},
           State#q{message_buffer = queue:new()});
 
-handle_call({claim_queue, ReaderPid}, _From,
-            State = #q{owner = Owner, exclusive_consumer = Holder}) ->
-    case Owner of
-        none ->
-            case check_exclusive_access(Holder, true, State) of
-                in_use ->
-                    %% FIXME: Is this really the right answer? What if
-                    %% an active consumer's reader is actually the
-                    %% claiming pid? Should that be allowed? In order
-                    %% to check, we'd need to hold not just the ch
-                    %% pid for each consumer, but also its reader
-                    %% pid...
-                    reply(locked, State);
-                ok ->
-                    MonitorRef = erlang:monitor(process, ReaderPid),
-                    reply(ok, State#q{owner = {ReaderPid, MonitorRef}})
-            end;
-        {ReaderPid, _MonitorRef} ->
+handle_call({requeue, MsgIds, ChPid}, _From, State) ->
+    case lookup_ch(ChPid) of
+        not_found ->
+            rabbit_log:warning("Ignoring requeue from unknown ch: ~p~n",
+                               [ChPid]),
             reply(ok, State);
-        _ ->
-            reply(locked, State)
+        C = #cr{unacked_messages = UAM} ->
+            {Messages, NewUAM} = collect_messages(MsgIds, UAM),
+            store_ch_record(C#cr{unacked_messages = NewUAM}),
+            reply(ok, deliver_or_enqueue_n(
+                        [{Message, true} || Message <- Messages], State))
     end.
 
+
 handle_cast({deliver, Txn, Message, ChPid}, State) ->
     %% Asynchronous, non-"mandatory", non-"immediate" deliver mode.
     {_Delivered, NewState} = deliver_or_enqueue(Txn, ChPid, Message, State),
@@ -787,19 +770,6 @@ handle_cast({rollback, Txn}, State) ->
 handle_cast({redeliver, Messages}, State) ->
     noreply(deliver_or_enqueue_n(Messages, State));
 
-handle_cast({requeue, MsgIds, ChPid}, State) ->
-    case lookup_ch(ChPid) of
-        not_found ->
-            rabbit_log:warning("Ignoring requeue from unknown ch: ~p~n",
-                               [ChPid]),
-            noreply(State);
-        C = #cr{unacked_messages = UAM} ->
-            {Messages, NewUAM} = collect_messages(MsgIds, UAM),
-            store_ch_record(C#cr{unacked_messages = NewUAM}),
-            noreply(deliver_or_enqueue_n(
-                      [{Message, true} || Message <- Messages], State))
-    end;
-
 handle_cast({unblock, ChPid}, State) ->
     noreply(
       possibly_unblock(State, ChPid,
@@ -828,19 +798,10 @@ handle_cast({limit, ChPid, LimiterPid}, State) ->
                 C#cr{limiter_pid = LimiterPid, is_limit_active = NewLimited}
         end)).
 
-handle_info({'DOWN', MonitorRef, process, DownPid, _Reason},
-            State = #q{owner = {DownPid, MonitorRef}}) ->
-    %% We know here that there are no consumers on this queue that are
-    %% owned by other pids than the one that just went down, so since
-    %% exclusive in some sense implies autodelete, we delete the queue
-    %% here. The other way of implementing the "exclusive implies
-    %% autodelete" feature is to actually set autodelete when an
-    %% exclusive declaration is seen, but this has the problem that
-    %% the python tests rely on the queue not going away after a
-    %% basic.cancel when the queue was declared exclusive and
-    %% nonautodelete.
-    NewState = State#q{owner = none},
-    {stop, normal, NewState};
+handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason},
+            State = #q{q= #amqqueue{ exclusive_owner = DownPid}}) ->
+    %% Exclusively owned queues must disappear with their owner.
+    {stop, normal, State};
 handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) ->
     case handle_ch_down(DownPid, State) of
         {ok, NewState}   -> noreply(NewState);
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 585c59dc..ddb941cf 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -292,6 +292,15 @@ check_write_permitted(Resource, #ch{ username = Username}) ->
 check_read_permitted(Resource, #ch{ username = Username}) ->
     check_resource_access(Username, Resource, read).
 
+check_queue_exclusivity(ReaderPid, Q) ->
+    case Q of
+        #amqqueue{ exclusive_owner = none} -> Q;
+        #amqqueue{ exclusive_owner = ReaderPid } -> Q;
+        _ -> rabbit_misc:protocol_error(resource_locked,
+                                        "cannot obtain exclusive access to locked ~s",
+                                        [rabbit_misc:rs(Q#amqqueue.name)])
+    end.
+
 expand_queue_name_shortcut(<<>>, #ch{ most_recently_declared_queue = <<>> }) ->
     rabbit_misc:protocol_error(
       not_allowed, "no previously declared queue", []);
@@ -346,9 +355,6 @@ handle_method(#'channel.close'{}, _, State = #ch{writer_pid = WriterPid}) ->
     ok = rabbit_writer:send_command(WriterPid, #'channel.close_ok'{}),
     stop;
 
-handle_method(#'access.request'{},_, State) ->
-    {reply, #'access.request_ok'{ticket = 1}, State};
-
 handle_method(#'basic.publish'{exchange = ExchangeNameBin,
                                routing_key = RoutingKey,
                                mandatory = Mandatory,
@@ -417,12 +423,16 @@ 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 }) ->
     QueueName = expand_queue_name_shortcut(QueueNameBin, State),
     check_read_permitted(QueueName, State),
     case rabbit_amqqueue:with_or_die(
            QueueName,
-           fun (Q) -> rabbit_amqqueue:basic_get(Q, self(), NoAck) end) of
+           fun (Q) ->
+                   check_queue_exclusivity(ReaderPid, Q),
+                   rabbit_amqqueue:basic_get(Q, self(), NoAck)
+           end) of
         {ok, MessageCount,
          Msg = {_QName, _QPid, _MsgId, Redelivered,
                 #basic_message{exchange_name = ExchangeName,
@@ -439,7 +449,7 @@ handle_method(#'basic.get'{queue = QueueNameBin,
                    Content),
             {noreply, State1#ch{next_tag = DeliveryTag + 1}};
         empty ->
-            {reply, #'basic.get_empty'{cluster_id = <<>>}, State}
+            {reply, #'basic.get_empty'{deprecated_cluster_id = <<>>}, State}
     end;
 
 handle_method(#'basic.consume'{queue = QueueNameBin,
@@ -467,6 +477,7 @@ handle_method(#'basic.consume'{queue = QueueNameBin,
             case rabbit_amqqueue:with_or_die(
                    QueueName,
                    fun (Q) ->
+                           check_queue_exclusivity(ReaderPid, Q),
                            rabbit_amqqueue:basic_consume(
                              Q, NoAck, ReaderPid, self(), LimiterPid,
                              ActualConsumerTag, ExclusiveConsume,
@@ -478,14 +489,6 @@ handle_method(#'basic.consume'{queue = QueueNameBin,
                                        dict:store(ActualConsumerTag,
                                                   QueueName,
                                                   ConsumerMapping)}};
-                {error, queue_owned_by_another_connection} ->
-                    %% The spec is silent on which exception to use
-                    %% here. This seems reasonable?
-                    %% FIXME: check this
-
-                    rabbit_misc:protocol_error(
-                      resource_locked, "~s owned by another connection",
-                      [rabbit_misc:rs(QueueName)]);
                 {error, exclusive_consume_unavailable} ->
                     rabbit_misc:protocol_error(
                       access_refused, "~s in exclusive use",
@@ -560,7 +563,7 @@ handle_method(#'basic.qos'{prefetch_count = PrefetchCount},
     ok = rabbit_limiter:limit(NewLimiterPid, PrefetchCount),
     {reply, #'basic.qos_ok'{}, State#ch{limiter_pid = NewLimiterPid}};
 
-handle_method(#'basic.recover'{requeue = true},
+handle_method(#'basic.recover_async'{requeue = true},
               _, State = #ch{ transaction_id = none,
                               unacked_message_q = UAMQ }) ->
     ok = fold_per_queue(
@@ -572,10 +575,11 @@ handle_method(#'basic.recover'{requeue = true},
                    rabbit_amqqueue:requeue(
                      QPid, lists:reverse(MsgIds), self())
            end, ok, UAMQ),
-    %% No answer required, apparently!
+    %% No answer required - basic.recover is the newer, synchronous
+    %% variant of this method
     {noreply, State#ch{unacked_message_q = queue:new()}};
 
-handle_method(#'basic.recover'{requeue = false},
+handle_method(#'basic.recover_async'{requeue = false},
               _, State = #ch{ transaction_id = none,
                               writer_pid = WriterPid,
                               unacked_message_q = UAMQ }) ->
@@ -596,20 +600,29 @@ handle_method(#'basic.recover'{requeue = false},
                    internal_deliver(
                      WriterPid, false, ConsumerTag, DeliveryTag,
                      {QName, QPid, MsgId, true, Message})
-           end, ok, UAMQ),
-    %% No answer required, apparently!
+           end, queue:to_list(UAMQ)),
+    %% No answer required - basic.recover is the newer, synchronous
+    %% variant of this method
     {noreply, State};
 
-handle_method(#'basic.recover'{}, _, _State) ->
+handle_method(#'basic.recover_async'{}, _, _State) ->
     rabbit_misc:protocol_error(
       not_allowed, "attempt to recover a transactional channel",[]);
 
+handle_method(#'basic.recover'{requeue = Requeue}, Content, State) ->
+    {noreply, State2 = #ch{writer_pid = WriterPid}} =
+        handle_method(#'basic.recover_async'{requeue = Requeue},
+                      Content,
+                      State),
+    ok = rabbit_writer:send_command(WriterPid, #'basic.recover_ok'{}),
+    {noreply, State2};
+
 handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
                                   type = TypeNameBin,
                                   passive = false,
                                   durable = Durable,
-                                  auto_delete = AutoDelete,
-                                  internal = false,
+                                  deprecated_auto_delete = false, %% 0-9-1: true not supported
+                                  deprecated_internal = false, %% 0-9-1: true not supported
                                   nowait = NoWait,
                                   arguments = Args},
               _, State = #ch{ virtual_host = VHostPath }) ->
@@ -630,21 +643,18 @@ handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
                 rabbit_exchange:declare(ExchangeName,
                                         CheckedType,
                                         Durable,
-                                        AutoDelete,
                                         Args)
         end,
-    ok = rabbit_exchange:assert_type(X, CheckedType),
+    ok = rabbit_exchange:assert_equivalence(X, CheckedType, Durable, Args),
     return_ok(State, NoWait, #'exchange.declare_ok'{});
 
 handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
-                                  type = TypeNameBin,
                                   passive = true,
                                   nowait = NoWait},
               _, State = #ch{ virtual_host = VHostPath }) ->
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_configure_permitted(ExchangeName, State),
-    X = rabbit_exchange:lookup_or_die(ExchangeName),
-    ok = rabbit_exchange:assert_type(X, rabbit_exchange:check_type(TypeNameBin)),
+    _ = rabbit_exchange:lookup_or_die(ExchangeName),
     return_ok(State, NoWait, #'exchange.declare_ok'{});
 
 handle_method(#'exchange.delete'{exchange = ExchangeNameBin,
@@ -672,25 +682,37 @@ handle_method(#'queue.declare'{queue = QueueNameBin,
                                arguments = Args},
               _, State = #ch { virtual_host = VHostPath,
                                reader_pid = ReaderPid }) ->
-    %% FIXME: atomic create&claim
-    Finish =
-        fun (Q) ->
-                if ExclusiveDeclare ->
-                        case rabbit_amqqueue:claim_queue(Q, ReaderPid) of
-                            locked ->
-                                %% AMQP 0-8 doesn't say which
-                                %% exception to use, so we mimic QPid
-                                %% here.
-                                rabbit_misc:protocol_error(
-                                  resource_locked,
-                                  "cannot obtain exclusive access to locked ~s",
-                                  [rabbit_misc:rs(Q#amqqueue.name)]);
-                            ok -> ok
-                        end;
-                   true ->
-                        ok
-                end,
-                Q
+    Owner = case ExclusiveDeclare of
+                true  -> ReaderPid;
+                false -> none
+            end,
+    %% We use this in both branches, because queue_declare may yet return an
+    %% existing queue.
+    Finish = 
+        fun(Q) ->
+                case Q of 
+                    %% "equivalent" rule. NB: we don't pay attention to
+                    %% anything in the arguments table, so for the sake of the
+                    %% "equivalent" rule, all tables of arguments are
+                    %% semantically equivalant.
+                    Matched = #amqqueue{name = QueueName,
+                                        durable = Durable, %% i.e., as supplied
+                                        exclusive_owner = Owner,
+                                        auto_delete = AutoDelete %% i.e,. as supplied
+                                       } ->
+                        check_configure_permitted(QueueName, State),
+                        Matched;
+                    %% exclusivity trumps non-equivalence arbitrarily
+                    #amqqueue{name = QueueName, exclusive_owner = ExclusiveOwner}
+                    when ExclusiveOwner =/= Owner ->
+                        rabbit_misc:protocol_error(resource_locked,
+                                                   "cannot obtain exclusive access to locked ~s",
+                                                   [rabbit_misc:rs(QueueName)]);
+                    #amqqueue{name = QueueName} ->                 
+                        rabbit_misc:protocol_error(channel_error,
+                                                   "parameters for ~s not equivalent",
+                                                   [rabbit_misc:rs(QueueName)])
+                    end
         end,
     Q = case rabbit_amqqueue:with(
                rabbit_misc:r(VHostPath, queue, QueueNameBin),
@@ -703,21 +725,20 @@ handle_method(#'queue.declare'{queue = QueueNameBin,
                     end,
                 QueueName = rabbit_misc:r(VHostPath, queue, ActualNameBin),
                 check_configure_permitted(QueueName, State),
-                Finish(rabbit_amqqueue:declare(QueueName,
-                                               Durable, AutoDelete, Args));
-            Other = #amqqueue{name = QueueName} ->
-                check_configure_permitted(QueueName, State),
-                Other
+                Finish(rabbit_amqqueue:declare(QueueName, Durable, AutoDelete, Args, Owner));
+            Found -> Found
         end,
     return_queue_declare_ok(State, NoWait, Q);
 
 handle_method(#'queue.declare'{queue = QueueNameBin,
                                passive = true,
                                nowait = NoWait},
-              _, State = #ch{ virtual_host = VHostPath }) ->
+              _, State = #ch{ virtual_host = VHostPath,
+                              reader_pid = ReaderPid }) ->
     QueueName = rabbit_misc:r(VHostPath, queue, QueueNameBin),
     check_configure_permitted(QueueName, State),
-    Q = rabbit_amqqueue:with_or_die(QueueName, fun (Q) -> Q end),
+    CheckExclusive = fun(Q) -> check_queue_exclusivity(ReaderPid, Q) end,
+    Q = rabbit_amqqueue:with_or_die(QueueName, CheckExclusive),
     return_queue_declare_ok(State, NoWait, Q);
 
 handle_method(#'queue.delete'{queue = QueueNameBin,
@@ -725,12 +746,15 @@ handle_method(#'queue.delete'{queue = QueueNameBin,
                               if_empty = IfEmpty,
                               nowait = NoWait
                              },
-              _, State) ->
+              _, State = #ch{ reader_pid = ReaderPid }) ->
     QueueName = expand_queue_name_shortcut(QueueNameBin, State),
     check_configure_permitted(QueueName, State),
     case rabbit_amqqueue:with_or_die(
            QueueName,
-           fun (Q) -> rabbit_amqqueue:delete(Q, IfUnused, IfEmpty) end) of
+           fun (Q) ->
+                   check_queue_exclusivity(ReaderPid, Q),
+                   rabbit_amqqueue:delete(Q, IfUnused, IfEmpty)
+           end) of
         {error, in_use} ->
             rabbit_misc:protocol_error(
               precondition_failed, "~s in use", [rabbit_misc:rs(QueueName)]);
@@ -748,7 +772,7 @@ handle_method(#'queue.bind'{queue = QueueNameBin,
                             routing_key = RoutingKey,
                             nowait = NoWait,
                             arguments = Arguments}, _, State) ->
-    binding_action(fun rabbit_exchange:add_binding/4, ExchangeNameBin,
+    binding_action(fun rabbit_exchange:add_binding/5, ExchangeNameBin,
                    QueueNameBin, RoutingKey, Arguments, #'queue.bind_ok'{},
                    NoWait, State);
 
@@ -756,18 +780,21 @@ handle_method(#'queue.unbind'{queue = QueueNameBin,
                               exchange = ExchangeNameBin,
                               routing_key = RoutingKey,
                               arguments = Arguments}, _, State) ->
-    binding_action(fun rabbit_exchange:delete_binding/4, ExchangeNameBin,
+    binding_action(fun rabbit_exchange:delete_binding/5, ExchangeNameBin,
                    QueueNameBin, RoutingKey, Arguments, #'queue.unbind_ok'{},
                    false, State);
 
 handle_method(#'queue.purge'{queue = QueueNameBin,
                              nowait = NoWait},
-              _, State) ->
+              _, State = #ch{ reader_pid = ReaderPid }) ->
     QueueName = expand_queue_name_shortcut(QueueNameBin, State),
     check_read_permitted(QueueName, State),
     {ok, PurgedMessageCount} = rabbit_amqqueue:with_or_die(
                                  QueueName,
-                                 fun (Q) -> rabbit_amqqueue:purge(Q) end),
+                                 fun (Q) ->
+                                         check_queue_exclusivity(ReaderPid, Q),
+                                         rabbit_amqqueue:purge(Q)
+                                 end),
     return_ok(State, NoWait,
               #'queue.purge_ok'{message_count = PurgedMessageCount});
 
@@ -808,7 +835,9 @@ handle_method(_MethodRecord, _Content, _State) ->
 %%----------------------------------------------------------------------------
 
 binding_action(Fun, ExchangeNameBin, QueueNameBin, RoutingKey, Arguments,
-               ReturnMethod, NoWait, State = #ch{virtual_host = VHostPath}) ->
+               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 -
@@ -819,7 +848,8 @@ binding_action(Fun, ExchangeNameBin, QueueNameBin, RoutingKey, Arguments,
                                                    State),
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_read_permitted(ExchangeName, State),
-    case Fun(ExchangeName, QueueName, ActualRoutingKey, Arguments) of
+    CheckExclusive = fun(_X, Q) -> check_queue_exclusivity(ReaderPid, Q) end,
+    case Fun(ExchangeName, QueueName, ActualRoutingKey, Arguments, CheckExclusive) of
         {error, exchange_not_found} ->
             rabbit_misc:not_found(ExchangeName);
         {error, queue_not_found} ->
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 6aac4428..2c6bd5ae 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -187,7 +187,7 @@ transactions, memory]. The default is to display name and (number of)
 messages.
 
 <ExchangeInfoItem> must be a member of the list [name, type, durable,
-auto_delete, arguments]. The default is to display name and type.
+arguments]. The default is to display name and type.
 
 The output format for \"list_bindings\" is a list of rows containing
 exchange name, queue name, routing key and arguments, in that order.
diff --git a/src/rabbit_error_logger.erl b/src/rabbit_error_logger.erl
index e9baf2c4..face0a1a 100644
--- a/src/rabbit_error_logger.erl
+++ b/src/rabbit_error_logger.erl
@@ -48,7 +48,7 @@ boot() ->
 init([DefaultVHost]) ->
     #exchange{} = rabbit_exchange:declare(
                     rabbit_misc:r(DefaultVHost, exchange, ?LOG_EXCH_NAME),
-                    topic, true, false, []),
+                    topic, true, []),
     {ok, #resource{virtual_host = DefaultVHost,
                    kind = exchange,
                    name = ?LOG_EXCH_NAME}}.
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 832acd16..bcba7d0b 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -34,12 +34,13 @@
 -include("rabbit.hrl").
 -include("rabbit_framing.hrl").
 
--export([recover/0, declare/5, lookup/1, lookup_or_die/1,
+-export([recover/0, declare/4, lookup/1, lookup_or_die/1,
          list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2,
          publish/2]).
--export([add_binding/4, delete_binding/4, list_bindings/1]).
+-export([add_binding/5, delete_binding/5, list_bindings/1]).
 -export([delete/2]).
 -export([delete_queue_bindings/1, delete_transient_queue_bindings/1]).
+-export([assert_equivalence/4]).
 -export([check_type/1, assert_type/2, topic_matches/2, headers_match/2]).
 
 %% EXTENDED API
@@ -60,11 +61,12 @@
                             'queue_not_found' |
                             'exchange_not_found' |
                             'exchange_and_queue_not_found'}).
+-type(inner_fun() :: fun((exchange(), queue()) -> any())).
+
 -spec(recover/0 :: () -> 'ok').
--spec(declare/5 :: (exchange_name(), exchange_type(), boolean(), boolean(),
-                    amqp_table()) -> exchange()).
+-spec(declare/4 :: (exchange_name(), exchange_type(), boolean(), amqp_table()) -> exchange()).
 -spec(check_type/1 :: (binary()) -> atom()).
--spec(assert_type/2 :: (exchange(), atom()) -> 'ok').
+-spec(assert_equivalence/4 :: (exchange(), atom(), boolean(), amqp_table()) -> 'ok').
 -spec(lookup/1 :: (exchange_name()) -> {'ok', exchange()} | not_found()).
 -spec(lookup_or_die/1 :: (exchange_name()) -> exchange()).
 -spec(list/1 :: (vhost()) -> [exchange()]).
@@ -74,11 +76,11 @@
 -spec(info_all/1 :: (vhost()) -> [[info()]]).
 -spec(info_all/2 :: (vhost(), [info_key()]) -> [[info()]]).
 -spec(publish/2 :: (exchange(), delivery()) -> {routing_result(), [pid()]}).
--spec(add_binding/4 ::
-      (exchange_name(), queue_name(), routing_key(), amqp_table()) ->
+-spec(add_binding/5 ::
+      (exchange_name(), queue_name(), routing_key(), amqp_table(), inner_fun()) ->
              bind_res() | {'error', 'durability_settings_incompatible'}).
--spec(delete_binding/4 ::
-      (exchange_name(), queue_name(), routing_key(), amqp_table()) ->
+-spec(delete_binding/5 ::
+      (exchange_name(), queue_name(), routing_key(), amqp_table(), inner_fun()) ->
              bind_res() | {'error', 'binding_not_found'}).
 -spec(list_bindings/1 :: (vhost()) ->
              [{exchange_name(), queue_name(), routing_key(), amqp_table()}]).
@@ -97,7 +99,7 @@
 
 %%----------------------------------------------------------------------------
 
--define(INFO_KEYS, [name, type, durable, auto_delete, arguments].
+-define(INFO_KEYS, [name, type, durable, arguments].
 
 recover() ->
     ok = rabbit_misc:table_foreach(
@@ -112,11 +114,10 @@ recover() ->
                                            ReverseRoute, write)
            end, rabbit_durable_route).
 
-declare(ExchangeName, Type, Durable, AutoDelete, Args) ->
+declare(ExchangeName, Type, Durable, Args) ->
     Exchange = #exchange{name = ExchangeName,
                          type = Type,
                          durable = Durable,
-                         auto_delete = AutoDelete,
                          arguments = Args},
     rabbit_misc:execute_mnesia_transaction(
       fun () ->
@@ -144,6 +145,16 @@ check_type(T) ->
     rabbit_misc:protocol_error(
       command_invalid, "invalid exchange type '~s'", [T]).
 
+assert_equivalence(X = #exchange{ durable = ActualDurable },
+                   RequiredType, RequiredDurable, RequiredArgs)
+  when ActualDurable==RequiredDurable ->
+    ok = assert_type(X, RequiredType),
+    ok = assert_args_equivalence(X, RequiredArgs);
+assert_equivalence(#exchange{ name = Name }, _Type, _Durable, _Args) ->
+    rabbit_misc:protocol_error(
+      not_allowed, "cannot redeclare ~s with different durable value",
+      [rabbit_misc:rs(Name)]).
+
 assert_type(#exchange{ type = ActualType }, RequiredType)
   when ActualType == RequiredType ->
     ok;
@@ -152,6 +163,24 @@ assert_type(#exchange{ name = Name, type = ActualType }, RequiredType) ->
       not_allowed, "cannot redeclare ~s of type '~s' with type '~s'",
       [rabbit_misc:rs(Name), ActualType, RequiredType]).
 
+alternate_exchange_value(Args) ->
+    lists:keysearch(<<"alternate-exchange">>, 1, Args).
+
+assert_args_equivalence(#exchange{ name = Name, 
+                                   arguments = Args },
+                        RequiredArgs) ->
+    %% The spec says "Arguments are compared for semantic
+    %% equivalence".  The only arg we care about is
+    %% "alternate-exchange".
+    Ae1 = alternate_exchange_value(RequiredArgs),
+    Ae2 = alternate_exchange_value(Args),
+    if Ae1==Ae2 -> ok; 
+          true  -> rabbit_misc:protocol_error(
+                     not_allowed,
+                     "cannot redeclare ~s with inequivalent args",
+                     [rabbit_misc:rs(Name)])
+    end.
+
 lookup(Name) ->
     rabbit_misc:dirty_read({rabbit_exchange, Name}).
 
@@ -178,7 +207,6 @@ infos(Items, X) -> [{Item, i(Item, X)} || Item <- Items].
 i(name,        #exchange{name        = Name})       -> Name;
 i(type,        #exchange{type        = Type})       -> Type;
 i(durable,     #exchange{durable     = Durable})    -> Durable;
-i(auto_delete, #exchange{auto_delete = AutoDelete}) -> AutoDelete;
 i(arguments,   #exchange{arguments   = Arguments})  -> Arguments;
 i(Item, _) -> throw({bad_argument, Item}).
 
@@ -327,10 +355,6 @@ delete_queue_bindings(QueueName, FwdDeleteFun) ->
                          #route{binding = #binding{queue_name = QueueName,
                                                    _ = '_'}}),
                        write)],
-    [begin
-         [X] = mnesia:read({rabbit_exchange, ExchangeName}),
-         ok = maybe_auto_delete(X)
-     end || ExchangeName <- Exchanges],
     ok.
 
 delete_forward_routes(Route) ->
@@ -384,27 +408,26 @@ call_with_exchange_and_queue(Exchange, Queue, Fun) ->
                end
       end).
 
-add_binding(ExchangeName, QueueName, RoutingKey, Arguments) ->
+add_binding(ExchangeName, QueueName, RoutingKey, Arguments, InnerFun) ->
     binding_action(
       ExchangeName, QueueName, RoutingKey, Arguments,
-      fun (X, Q, B) ->
-              if Q#amqqueue.durable and not(X#exchange.durable) ->
-                      {error, durability_settings_incompatible};
-                 true -> ok = sync_binding(B, Q#amqqueue.durable,
-                                           fun mnesia:write/3)
-              end
+      fun (X = #exchange{type = Type}, Q, B) ->
+              InnerFun(X, Q),
+              ok = sync_binding(B, Q#amqqueue.durable,
+                                fun mnesia:write/3)
       end).
 
-delete_binding(ExchangeName, QueueName, RoutingKey, Arguments) ->
+delete_binding(ExchangeName, QueueName, RoutingKey, Arguments, InnerFun) ->
     binding_action(
       ExchangeName, QueueName, RoutingKey, Arguments,
-      fun (X, Q, B) ->
+      fun (X = #exchange{type = Type}, Q, B) ->
               case mnesia:match_object(rabbit_route, #route{binding = B},
                                        write) of
                   [] -> {error, binding_not_found};
-                  _  -> ok = sync_binding(B, Q#amqqueue.durable,
-                                          fun mnesia:delete_object/3),
-                        maybe_auto_delete(X)
+                  _  ->
+                      InnerFun(X, Q),
+                      ok = sync_binding(B, Q#amqqueue.durable,
+                                        fun mnesia:delete_object/3)
               end
       end).
 
@@ -566,12 +589,6 @@ delete(ExchangeName, _IfUnused = true) ->
 delete(ExchangeName, _IfUnused = false) ->
     call_with_exchange(ExchangeName, fun unconditional_delete/1).
 
-maybe_auto_delete(#exchange{auto_delete = false}) ->
-    ok;
-maybe_auto_delete(Exchange = #exchange{auto_delete = true}) ->
-    conditional_delete(Exchange),
-    ok.
-
 conditional_delete(Exchange = #exchange{name = ExchangeName}) ->
     Match = #route{binding = #binding{exchange_name = ExchangeName, _ = '_'}},
     %% we need to check for durable routes here too in case a bunch of
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 1a4830e1..738f7e3f 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -50,6 +50,9 @@
 -define(NORMAL_TIMEOUT, 3).
 -define(CLOSING_TIMEOUT, 1).
 -define(CHANNEL_TERMINATION_TIMEOUT, 3).
+%% set to zero once QPid fix their negotiation
+-define(FRAME_MAX, 131072).
+-define(CHANNEL_MAX, 0).
 
 %---------------------------------------------------------------------------
 
@@ -436,7 +439,6 @@ handle_frame(Type, 0, Payload, State) ->
     case analyze_frame(Type, Payload) of
         error     -> throw({unknown_frame, 0, Type, Payload});
         heartbeat -> State;
-        trace     -> State;
         {method, MethodName, FieldsBin} ->
             handle_method0(MethodName, FieldsBin, State);
         Other -> throw({unexpected_frame_on_channel0, Other})
@@ -445,7 +447,6 @@ handle_frame(Type, Channel, Payload, State) ->
     case analyze_frame(Type, Payload) of
         error         -> throw({unknown_frame, Channel, Type, Payload});
         heartbeat     -> throw({unexpected_heartbeat_frame, Channel});
-        trace         -> throw({unexpected_trace_frame, Channel});
         AnalyzedFrame ->
             %%?LOGDEBUG("Ch ~p Frame ~p~n", [Channel, AnalyzedFrame]),
             case get({channel, Channel}) of
@@ -484,8 +485,6 @@ analyze_frame(?FRAME_HEADER, <<ClassId:16, Weight:16, BodySize:64, Properties/bi
     {content_header, ClassId, Weight, BodySize, Properties};
 analyze_frame(?FRAME_BODY, Body) ->
     {content_body, Body};
-analyze_frame(?FRAME_TRACE, _Body) ->
-    trace;
 analyze_frame(?FRAME_HEARTBEAT, <<>>) ->
     heartbeat;
 analyze_frame(_Type, _Body) ->
@@ -505,56 +504,58 @@ handle_input({frame_payload, Type, Channel, PayloadSize}, PayloadAndMarker, Stat
             throw({bad_payload, PayloadAndMarker})
     end;
 
-handle_input(handshake, <<"AMQP",1,1,ProtocolMajor,ProtocolMinor>>,
-             State = #v1{sock = Sock, connection = Connection}) ->
-    case check_version({ProtocolMajor, ProtocolMinor},
-                       {?PROTOCOL_VERSION_MAJOR, ?PROTOCOL_VERSION_MINOR}) of
-        true ->
-            {ok, Product} = application:get_key(id),
-            {ok, Version} = application:get_key(vsn),
-            ok = send_on_channel0(
-                   Sock,
-                   #'connection.start'{
-                     version_major = ?PROTOCOL_VERSION_MAJOR,
-                     version_minor = ?PROTOCOL_VERSION_MINOR,
-                     server_properties =
-                     [{list_to_binary(K), longstr, list_to_binary(V)} ||
-                         {K, V} <-
-                             [{"product",     Product},
-                              {"version",     Version},
-                              {"platform",    "Erlang/OTP"},
-                              {"copyright",   ?COPYRIGHT_MESSAGE},
-                              {"information", ?INFORMATION_MESSAGE}]],
-                     mechanisms = <<"PLAIN AMQPLAIN">>,
-                     locales = <<"en_US">> }),
-            {State#v1{connection = Connection#connection{
-                                     timeout_sec = ?NORMAL_TIMEOUT},
-                      connection_state = starting},
-             frame_header, 7};
-        false ->
-            throw({bad_version, ProtocolMajor, ProtocolMinor})
-    end;
-
+%% The two rules pertaining to version negotiation:
+%%
+%% * If the server cannot support the protocol specified in the
+%% protocol header, it MUST respond with a valid protocol header and
+%% then close the socket connection.
+%%
+%% * The server MUST provide a protocol version that is lower than or
+%% equal to that requested by the client in the protocol header.
+%%
+%% We support 0-9-1 and 0-9, so by the first rule, we must close the
+%% connection if we're sent anything else.  Then, we must send that
+%% version in the Connection.start method.
+handle_input(handshake, <<"AMQP",0,0,9,1>>, State) ->
+    %% 0-9-1 style protocol header.
+    protocol_negotiate(0, 9, 1, State);
+handle_input(handshake, <<"AMQP",1,1,0,9>>, State) ->
+    %% 0-8 and 0-9 style protocol header; we support only 0-9
+    protocol_negotiate(0, 9, 0, State);
 handle_input(handshake, Other, #v1{sock = Sock}) ->
     ok = inet_op(fun () -> rabbit_net:send(
-                             Sock, <<"AMQP",1,1,
-                                    ?PROTOCOL_VERSION_MAJOR,
-                                    ?PROTOCOL_VERSION_MINOR>>) end),
+                             Sock, <<"AMQP",0,0,9,1>>) end),
     throw({bad_header, Other});
 
 handle_input(Callback, Data, _State) ->
     throw({bad_input, Callback, Data}).
 
-%% the 0-8 spec, confusingly, defines the version as 8-0
-adjust_version({8,0})   -> {0,8};
-adjust_version(Version) -> Version.
-check_version(ClientVersion, ServerVersion) ->
-    {ClientMajor, ClientMinor} = adjust_version(ClientVersion),
-    {ServerMajor, ServerMinor} = adjust_version(ServerVersion),
-    ClientMajor > ServerMajor
-        orelse
-          (ClientMajor == ServerMajor andalso
-           ClientMinor >= ServerMinor).
+%% Offer a protocol version to the client..  Connection.start only
+%% includes a major and minor version number, Luckily 0-9 and 0-9-1
+%% are similar enough that clients will be happy with either.
+protocol_negotiate(ProtocolMajor, ProtocolMinor, _ProtocolRevision,
+                   State = #v1{sock = Sock, connection = Connection}) ->
+    {ok, Product} = application:get_key(id),
+    {ok, Version} = application:get_key(vsn),
+    ok = send_on_channel0(
+           Sock,
+           #'connection.start'{
+             version_major = ProtocolMajor,
+             version_minor = ProtocolMinor,
+             server_properties =
+             [{list_to_binary(K), longstr, list_to_binary(V)} ||
+                 {K, V} <-
+                     [{"product",     Product},
+                      {"version",     Version},
+                      {"platform",    "Erlang/OTP"},
+                      {"copyright",   ?COPYRIGHT_MESSAGE},
+                      {"information", ?INFORMATION_MESSAGE}]],
+             mechanisms = <<"PLAIN AMQPLAIN">>,
+             locales = <<"en_US">> }),
+    {State#v1{connection = Connection#connection{
+                             timeout_sec = ?NORMAL_TIMEOUT},
+              connection_state = starting},
+     frame_header, 7}.
 
 %%--------------------------------------------------------------------------
 
@@ -584,9 +585,8 @@ handle_method0(#'connection.start_ok'{mechanism = Mechanism,
     User = rabbit_access_control:check_login(Mechanism, Response),
     ok = send_on_channel0(
            Sock,
-           #'connection.tune'{channel_max = 0,
-                              %% set to zero once QPid fix their negotiation
-                              frame_max = 131072,
+           #'connection.tune'{channel_max = ?CHANNEL_MAX,
+                              frame_max = ?FRAME_MAX,
                               heartbeat = 0}),
     State#v1{connection_state = tuning,
              connection = Connection#connection{
@@ -598,43 +598,35 @@ handle_method0(#'connection.tune_ok'{channel_max = _ChannelMax,
                State = #v1{connection_state = tuning,
                            connection = Connection,
                            sock = Sock}) ->
-    %% if we have a channel_max limit that the client wishes to
+    if (FrameMax =< ?FRAME_MIN_SIZE) or
+       (?FRAME_MAX /= 0) and (FrameMax > ?FRAME_MAX) ->
+            rabbit_misc:protocol_error(
+              mistuned, "peer sent tune_ok with invalid frame_max");
+    %% If we have a channel_max limit that the client wishes to
     %% exceed, die as per spec.  Not currently a problem, so we ignore
     %% the client's channel_max parameter.
-    rabbit_heartbeat:start_heartbeat(Sock, ClientHeartbeat),
-    State#v1{connection_state = opening,
-             connection = Connection#connection{
-                            timeout_sec = ClientHeartbeat,
-                            frame_max = FrameMax}};
-handle_method0(#'connection.open'{virtual_host = VHostPath,
-                                  insist = Insist},
+%%        (?CHANNEL_MAX /= 0) and (ChannelMax > ?CHANNEL_MAX) ->
+%%             rabbit_misc:protocol_error(
+%%               mistuned, "peer sent tune_ok with invalid channel_max");
+       true ->
+            rabbit_heartbeat:start_heartbeat(Sock, ClientHeartbeat),
+            State#v1{connection_state = opening,
+                     connection = Connection#connection{
+                                    timeout_sec = ClientHeartbeat,
+                                    frame_max = FrameMax}}
+    end;
+handle_method0(#'connection.open'{virtual_host = VHostPath},
                State = #v1{connection_state = opening,
                            connection = Connection = #connection{
                                           user = User},
                            sock = Sock}) ->
     ok = rabbit_access_control:check_vhost_access(User, VHostPath),
     NewConnection = Connection#connection{vhost = VHostPath},
-    KnownHosts = format_listeners(rabbit_networking:active_listeners()),
-    Redirects = compute_redirects(Insist),
-    if Redirects == [] ->
-            ok = send_on_channel0(
-                   Sock,
-                   #'connection.open_ok'{known_hosts = KnownHosts}),
-            State#v1{connection_state = running,
-                     connection = NewConnection};
-       true ->
-            %% FIXME: 'host' is supposed to only contain one
-            %% address; but which one do we pick? This is
-            %% really a problem with the spec.
-            Host = format_listeners(Redirects),
-            rabbit_log:info("connection ~p redirecting to ~p~n",
-                            [self(), Host]),
-            ok = send_on_channel0(
-                   Sock,
-                   #'connection.redirect'{host = Host,
-                                          known_hosts = KnownHosts}),
-            close_connection(State#v1{connection = NewConnection})
-    end;
+    ok = send_on_channel0(
+           Sock,
+           #'connection.open_ok'{deprecated_known_hosts = <<>>}),
+    State#v1{connection_state = running,
+             connection = NewConnection};
 handle_method0(#'connection.close'{},
                State = #v1{connection_state = running}) ->
     lists:foreach(fun rabbit_framing_channel:shutdown/1, all_channels()),
@@ -653,21 +645,6 @@ handle_method0(_Method, #v1{connection_state = S}) ->
 send_on_channel0(Sock, Method) ->
     ok = rabbit_writer:internal_send_command(Sock, 0, Method).
 
-format_listeners(Listeners) ->
-    list_to_binary(
-      rabbit_misc:intersperse(
-        $,,
-        [io_lib:format("~s:~w", [Host, Port]) ||
-            #listener{host = Host, port = Port} <- Listeners])).
-
-compute_redirects(true) -> [];
-compute_redirects(false) ->
-    Node = node(),
-    LNode = rabbit_load:pick(),
-    if Node == LNode -> [];
-       true -> rabbit_networking:node_listeners(LNode)
-    end.
-
 %%--------------------------------------------------------------------------
 
 infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index 277e6717..8602a209 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -52,6 +52,7 @@ all_tests() ->
     passed = test_pg_local(),
     passed = test_unfold(),
     passed = test_parsing(),
+    passed = test_content_framing(),
     passed = test_topic_matching(),
     passed = test_log_management(),
     passed = test_app_management(),
@@ -321,6 +322,45 @@ test_field_values() ->
        >>),
     passed.
 
+%% Test that content frames don't exceed frame-max
+test_content_framing(FrameMax, Fragments) ->
+    [Header | Frames] =
+        rabbit_binary_generator:build_simple_content_frames(
+          1,
+          #content{class_id = 0, properties_bin = <<>>,
+                   payload_fragments_rev = Fragments},
+          FrameMax),
+    % header is formatted correctly and the size is the total of the
+    % fragments
+    <<_FrameHeader:7/binary, _ClassAndWeight:4/binary,
+     BodySize:64/unsigned, _Rest/binary>> = list_to_binary(Header),
+    BodySize = size(list_to_binary(Fragments)),
+    false = lists:any(
+              fun (ContentFrame) ->
+                      FrameBinary = list_to_binary(ContentFrame),
+                      % assert
+                      <<_TypeAndChannel:3/binary,
+                       Size:32/unsigned,
+                       _Payload:Size/binary,
+                       16#CE>> = FrameBinary,
+                      size(FrameBinary) > FrameMax
+              end,
+              Frames),
+    passed.
+
+test_content_framing() ->
+    % no content
+    passed = test_content_framing(4096, []),
+    passed = test_content_framing(4096, [<<>>]),
+    % easily fit in one frame
+    passed = test_content_framing(4096,   [<<"Easy">>]),
+    % exactly one frame (empty frame = 8 bytes)
+    passed = test_content_framing(11, [<<"One">>]),
+    % more than one frame
+    passed = test_content_framing(20, [<<"into more than one frame">>,
+                                       <<"This will have to go">>]),
+    passed.
+
 test_topic_match(P, R) ->
     test_topic_match(P, R, true).
 
@@ -711,19 +751,16 @@ test_user_management() ->
     passed.
 
 test_server_status() ->
-
     %% create a few things so there is some useful information to list
     Writer = spawn(fun () -> receive shutdown -> ok end end),
     Ch = rabbit_channel:start_link(1, self(), Writer, <<"user">>, <<"/">>),
     [Q, Q2] = [#amqqueue{} = rabbit_amqqueue:declare(
                                rabbit_misc:r(<<"/">>, queue, Name),
-                               false, false, []) ||
+                               false, false, [], none) ||
                   Name <- [<<"foo">>, <<"bar">>]],
 
-    ok = rabbit_amqqueue:claim_queue(Q, self()),
     ok = rabbit_amqqueue:basic_consume(Q, true, self(), Ch, undefined,
                                        <<"ctag">>, true, undefined),
-
     %% list queues
     ok = info_action(list_queues, rabbit_amqqueue:info_keys(), true),