diff options
Diffstat (limited to 'src/rabbit_binding.erl')
| -rw-r--r-- | src/rabbit_binding.erl | 124 |
1 files changed, 67 insertions, 57 deletions
diff --git a/src/rabbit_binding.erl b/src/rabbit_binding.erl index bb29580f..ec1816ca 100644 --- a/src/rabbit_binding.erl +++ b/src/rabbit_binding.erl @@ -33,11 +33,12 @@ -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_exchange/1, list_for_destination/1, + list_for_exchange_and_destination/2]). -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]). + remove_for_destination/1, remove_transient_for_destination/1]). %%---------------------------------------------------------------------------- @@ -66,9 +67,11 @@ 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_destination/1 :: + (rabbit_amqqueue:name()|rabbit_exchange:name()) -> bindings()). +-spec(list_for_exchange_and_destination/2 :: + (rabbit_exchange:name(), + rabbit_amqqueue:name() | rabbit_exchange:name()) -> 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()]) -> @@ -78,16 +81,16 @@ -> [[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(remove_for_destination/1 :: + (rabbit_amqqueue:name() | rabbit_exchange:name()) -> fun (() -> any())). +-spec(remove_transient_for_destination/1 :: + (rabbit_amqqueue:name() | rabbit_exchange:name()) -> fun (() -> any())). -endif. %%---------------------------------------------------------------------------- --define(INFO_KEYS, [exchange_name, queue_name, routing_key, arguments]). +-define(INFO_KEYS, [exchange_name, destination, routing_key, arguments]). recover() -> rabbit_misc:table_fold( @@ -101,26 +104,24 @@ recover() -> exists(Binding) -> binding_action( Binding, - fun (_X, _Q, B) -> mnesia:read({rabbit_route, B}) /= [] end). + fun (_X, _D, B) -> mnesia:read({rabbit_route, B}) /= [] end). -add(Binding) -> add(Binding, fun (_X, _Q) -> ok end). +add(Binding) -> add(Binding, fun (_X, _D) -> ok end). -remove(Binding) -> remove(Binding, fun (_X, _Q) -> ok end). +remove(Binding) -> remove(Binding, fun (_X, _D) -> ok end). add(Binding, InnerFun) -> case binding_action( Binding, - fun (X, Q, B) -> + fun (X, D, 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(X, D) 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, are_endpoints_durable(X, D), fun mnesia:write/3), {new, X, B}; [_] -> {existing, X, B} @@ -141,16 +142,14 @@ add(Binding, InnerFun) -> remove(Binding, InnerFun) -> case binding_action( Binding, - fun (X, Q, B) -> + fun (X, D, B) -> case mnesia:match_object(rabbit_route, #route{binding = B}, write) of [] -> {error, binding_not_found}; - [_] -> case InnerFun(X, Q) of + [_] -> case InnerFun(X, D) of ok -> - Durable = (X#exchange.durable andalso - Q#amqqueue.durable), ok = sync_binding( - B, Durable, + B, are_endpoints_durable(X, D), fun mnesia:delete_object/3), Deleted = rabbit_exchange:maybe_auto_delete(X), @@ -175,7 +174,7 @@ remove(Binding, InnerFun) -> list(VHostPath) -> Route = #route{binding = #binding{ exchange_name = rabbit_misc:r(VHostPath, exchange), - queue_name = rabbit_misc:r(VHostPath, queue), + destination = rabbit_misc:r(VHostPath, '_'), _ = '_'}, _ = '_'}, [B || #route{binding = B} <- mnesia:dirty_match_object(rabbit_route, @@ -186,15 +185,15 @@ list_for_exchange(ExchangeName) -> [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(DestinationName) -> + Route = #route{binding = #binding{destination = DestinationName, _ = '_'}}, [reverse_binding(B) || #reverse_route{reverse_binding = B} <- mnesia:dirty_match_object(rabbit_reverse_route, reverse_route(Route))]. -list_for_exchange_and_queue(ExchangeName, QueueName) -> +list_for_exchange_and_destination(ExchangeName, DestinationName) -> Route = #route{binding = #binding{exchange_name = ExchangeName, - queue_name = QueueName, + destination = DestinationName, _ = '_'}}, [B || #route{binding = B} <- mnesia:dirty_match_object(rabbit_route, Route)]. @@ -208,10 +207,10 @@ 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(exchange_name, #binding{exchange_name = XName}) -> XName; +i(destination, #binding{destination = Destination}) -> Destination; +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). @@ -241,22 +240,28 @@ remove_for_exchange(ExchangeName) -> _ = '_'}}, write)]. -remove_for_queue(QueueName) -> - remove_for_queue(QueueName, fun delete_forward_routes/1). +remove_for_destination(DestinationName) -> + remove_for_destination(DestinationName, fun delete_forward_routes/1). -remove_transient_for_queue(QueueName) -> - remove_for_queue(QueueName, fun delete_transient_forward_routes/1). +remove_transient_for_destination(DestinationName) -> + remove_for_destination(DestinationName, + fun delete_transient_forward_routes/1). %%---------------------------------------------------------------------------- +are_endpoints_durable(#exchange{durable = A}, #amqqueue{durable = B}) -> + A andalso B; +are_endpoints_durable(#exchange{durable = A}, #exchange{durable = B}) -> + A andalso B. + binding_action(Binding = #binding{exchange_name = ExchangeName, - queue_name = QueueName, + destination = Destination, args = Arguments}, Fun) -> - call_with_exchange_and_queue( - ExchangeName, QueueName, - fun (X, Q) -> + call_with_exchange_and_destination( + ExchangeName, Destination, + fun (X, D) -> SortedArgs = rabbit_misc:sort_field_table(Arguments), - Fun(X, Q, Binding#binding{args = SortedArgs}) + Fun(X, D, Binding#binding{args = SortedArgs}) end). sync_binding(Binding, Durable, Fun) -> @@ -270,15 +275,19 @@ sync_binding(Binding, Durable, Fun) -> ok = Fun(rabbit_reverse_route, ReverseRoute, write), ok. -call_with_exchange_and_queue(Exchange, Queue, Fun) -> +call_with_exchange_and_destination(Exchange, Destination, Fun) -> + DestTable = case Destination#resource.kind of + queue -> rabbit_queue; + exchange -> rabbit_exchange + end, rabbit_misc:execute_mnesia_transaction( fun () -> case {mnesia:read({rabbit_exchange, Exchange}), - mnesia:read({rabbit_queue, Queue})} of - {[X], [Q]} -> Fun(X, Q); - {[ ], [_]} -> {error, exchange_not_found}; - {[_], [ ]} -> {error, queue_not_found}; - {[ ], [ ]} -> {error, exchange_and_queue_not_found} - end + mnesia:read({DestTable, Destination})} of + {[X], [D]} -> Fun(X, D); + {[ ], [_]} -> {error, exchange_not_found}; + {[_], [ ]} -> {error, destination_not_found}; + {[ ], [ ]} -> {error, exchange_and_destination_not_found} + end end). %% Used with atoms from records; e.g., the type is expected to exist. @@ -293,7 +302,7 @@ continue('$end_of_table') -> false; continue({[_|_], _}) -> true; continue({[], Continuation}) -> continue(mnesia:select(Continuation)). -remove_for_queue(QueueName, FwdDeleteFun) -> +remove_for_destination(DestinationName, FwdDeleteFun) -> DeletedBindings = [begin Route = reverse_route(ReverseRoute), @@ -304,9 +313,10 @@ 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 = DestinationName, + _ = '_'}}), write)], Grouped = group_bindings_and_auto_delete( lists:keysort(#binding.exchange_name, DeletedBindings), []), @@ -360,19 +370,19 @@ reverse_route(#reverse_route{reverse_binding = Binding}) -> #route{binding = reverse_binding(Binding)}. reverse_binding(#reverse_binding{exchange_name = Exchange, - queue_name = Queue, + destination = Destination, key = Key, args = Args}) -> #binding{exchange_name = Exchange, - queue_name = Queue, + destination = Destination, key = Key, args = Args}; reverse_binding(#binding{exchange_name = Exchange, - queue_name = Queue, + destination = Destination, key = Key, args = Args}) -> #reverse_binding{exchange_name = Exchange, - queue_name = Queue, + destination = Destination, key = Key, args = Args}. |