Move exchange and queue equivalence semantics from amqp_0_9_1 to default

2 files changed, 54 insertions, 8 deletions

diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index f355bef3..1af07793 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -687,18 +687,17 @@ handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
                                         AutoDelete,
                                         Args)
         end,
-    ok = rabbit_exchange:assert_type(X, CheckedType),
+    ok = rabbit_exchange:assert_equivalence(X, CheckedType, Durable,
+                                            AutoDelete, 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,
@@ -733,14 +732,27 @@ handle_method(#'queue.declare'{queue       = QueueNameBin,
             end,
     %% We use this in both branches, because queue_declare may yet return an
     %% existing queue.
-    Finish = fun (#amqqueue{name = QueueName} = Q) ->
+    Finish = fun (#amqqueue{name = QueueName,
+                            durable = Durable1,
+                            auto_delete = AutoDelete1} = Q)
+                   when Durable =:= Durable1, AutoDelete =:= AutoDelete1 ->
                      check_exclusive_access(Q, Owner, strict),
                      check_configure_permitted(QueueName, State),
+                     %% We need to notify the reader within the channel
+                     %% process so that we can be sure there are no
+                     %% outstanding exclusive queues being declared as the
+                     %% connection shuts down.
                      case Owner of
                          none -> ok;
                          _    -> ok = rabbit_reader_queue_collector:register_exclusive_queue(CollectorPid, Q)
                      end,
-                     Q
+                     Q;
+                 %% non-equivalence trumps exclusivity arbitrarily
+                 (#amqqueue{name = QueueName}) ->
+                     rabbit_misc:protocol_error(
+                       channel_error,
+                       "parameters for ~s not equivalent",
+                       [rabbit_misc:rs(QueueName)])
              end,
     Q = case rabbit_amqqueue:with(
                rabbit_misc:r(VHostPath, queue, QueueNameBin),
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 99799f7a..a3deeb7b 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -39,7 +39,8 @@
 -export([add_binding/5, delete_binding/5, list_bindings/1]).
 -export([delete/2]).
 -export([delete_queue_bindings/1, delete_transient_queue_bindings/1]).
--export([check_type/1, assert_type/2]).
+-export([assert_equivalence/5]).
+-export([check_type/1]).
 
 %% EXTENDED API
 -export([list_exchange_bindings/1]).
@@ -64,7 +65,8 @@
 -spec(declare/5 :: (exchange_name(), exchange_type(), boolean(), boolean(),
                     amqp_table()) -> exchange()).
 -spec(check_type/1 :: (binary()) -> atom()).
--spec(assert_type/2 :: (exchange(), atom()) -> 'ok').
+-spec(assert_equivalence/5 :: (exchange(), atom(), boolean(), 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()]).
@@ -185,6 +187,20 @@ check_type(TypeBin) ->
             T
     end.
 
+assert_equivalence(X = #exchange{ durable = ActualDurable,
+                                  auto_delete = ActualAutoDelete},
+                   RequiredType, RequiredDurable, RequiredAutoDelete,
+                   RequiredArgs)
+  when ActualDurable =:= RequiredDurable andalso
+       ActualAutoDelete =:= RequiredAutoDelete ->
+    ok = assert_type(X, RequiredType),
+    ok = assert_args_equivalence(X, RequiredArgs);
+assert_equivalence(#exchange{ name = Name }, _Type, _Durable, _AutoDelete,
+                   _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;
@@ -193,6 +209,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}).