Merge pull request #50 from rabbitmq/rabbitmq-server-590

Pass connection Pid to exclusive queue deletion.

deps/rabbit_common/mk/rabbitmq-build.mk

@@ -2,7 +2,9 @@
 LAGER_EXTRA_SINKS += rabbit_log \
 		     rabbit_channel \
 		     rabbit_connection \
-		     rabbit_mirroring
+		     rabbit_mirroring \
+		     rabbit_queue \
+		     rabbit_federation
 lager_extra_sinks = $(subst $(space),$(comma),$(LAGER_EXTRA_SINKS))
 
 RMQ_ERLC_OPTS += +'{parse_transform,lager_transform}' \

deps/rabbit_common/src/rabbit_amqqueue.erl

@@ -17,8 +17,8 @@
 -module(rabbit_amqqueue).
 
 -export([recover/0, stop/0, start/1, declare/5, declare/6,
-         delete_immediately/1, delete/3, purge/1, forget_all_durable/1,
-         delete_crashed/1, delete_crashed_internal/1]).
+         delete_immediately/1, delete_exclusive/2, delete/3, purge/1, 
+         forget_all_durable/1, delete_crashed/1, delete_crashed_internal/1]).
 -export([pseudo_queue/2, immutable/1]).
 -export([lookup/1, not_found_or_absent/1, with/2, with/3, with_or_die/2,
          assert_equivalence/5,
@@ -139,6 +139,7 @@
         (rabbit_types:amqqueue())
         -> {'ok', non_neg_integer(), non_neg_integer()}).
 -spec(delete_immediately/1 :: (qpids()) -> 'ok').
+-spec(delete_exclusive/2 :: (qpids(), pid()) -> 'ok').
 -spec(delete/3 ::
         (rabbit_types:amqqueue(), 'false', 'false')
         -> qlen();
@@ -662,6 +663,10 @@ get_queue_consumer_info(Q, ConsumerInfoKeys) ->
 
 stat(#amqqueue{pid = QPid}) -> delegate:call(QPid, stat).
 
+delete_exclusive(QPids, ConnId) ->
+    [gen_server2:cast(QPid, {delete_exclusive, ConnId}) || QPid <- QPids],
+    ok.
+
 delete_immediately(QPids) ->
     [gen_server2:cast(QPid, delete_immediately) || QPid <- QPids],
     ok.

deps/rabbit_common/src/rabbit_queue_collector.erl

@@ -64,7 +64,7 @@ handle_call({register, QPid}, _From,
             State = #state{monitors = QMons, delete_from = Deleting}) ->
     case Deleting of
         undefined -> ok;
-        _         -> ok = rabbit_amqqueue:delete_immediately([QPid])
+        _         -> ok = rabbit_amqqueue:delete_exclusive([QPid], Deleting)
     end,
     {reply, ok, State#state{monitors = pmon:monitor(QPid, QMons)}};
 
@@ -72,7 +72,7 @@ handle_call(delete_all, From, State = #state{monitors    = QMons,
                                              delete_from = undefined}) ->
     case pmon:monitored(QMons) of
         []    -> {reply, ok, State#state{delete_from = From}};
-        QPids -> ok = rabbit_amqqueue:delete_immediately(QPids),
+        QPids -> ok = rabbit_amqqueue:delete_exclusive(QPids, From),
                  {noreply, State#state{delete_from = From}}
     end.