diff options
Diffstat (limited to 'src/rabbit_exchange.erl')
| -rw-r--r-- | src/rabbit_exchange.erl | 141 |
1 files changed, 83 insertions, 58 deletions
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl index 2fba941f..b4bdd348 100644 --- a/src/rabbit_exchange.erl +++ b/src/rabbit_exchange.erl @@ -22,7 +22,7 @@ assert_equivalence/6, assert_args_equivalence/2, check_type/1, lookup/1, lookup_or_die/1, list/1, lookup_scratch/2, update_scratch/3, info_keys/0, info/1, info/2, info_all/1, info_all/2, - route/2, delete/2]). + route/2, delete/2, validate_binding/2]). %% these must be run inside a mnesia tx -export([maybe_auto_delete/1, serial/1, peek_serial/1, update/2]). @@ -39,8 +39,7 @@ -spec(recover/0 :: () -> [name()]). -spec(callback/4:: (rabbit_types:exchange(), fun_name(), - fun((boolean()) -> non_neg_integer()) | atom(), - [any()]) -> 'ok'). + fun((boolean()) -> non_neg_integer()) | atom(), [any()]) -> 'ok'). -spec(policy_changed/2 :: (rabbit_types:exchange(), rabbit_types:exchange()) -> 'ok'). -spec(declare/6 :: @@ -69,7 +68,8 @@ -spec(update_scratch/3 :: (name(), atom(), fun((any()) -> any())) -> 'ok'). -spec(update/2 :: (name(), - fun((rabbit_types:exchange()) -> rabbit_types:exchange())) -> 'ok'). + fun((rabbit_types:exchange()) -> rabbit_types:exchange())) + -> not_found | rabbit_types:exchange()). -spec(info_keys/0 :: () -> rabbit_types:info_keys()). -spec(info/1 :: (rabbit_types:exchange()) -> rabbit_types:infos()). -spec(info/2 :: @@ -84,6 +84,9 @@ (name(), boolean())-> 'ok' | rabbit_types:error('not_found') | rabbit_types:error('in_use')). +-spec(validate_binding/2 :: + (rabbit_types:exchange(), rabbit_types:binding()) + -> rabbit_types:ok_or_error({'binding_invalid', string(), [any()]})). -spec(maybe_auto_delete/1:: (rabbit_types:exchange()) -> 'not_deleted' | {'deleted', rabbit_binding:deletions()}). @@ -111,29 +114,40 @@ recover() -> callback(X, create, map_create_tx(Tx), [X]) end, rabbit_durable_exchange), + report_missing_decorators(Xs), [XName || #exchange{name = XName} <- Xs]. -callback(X = #exchange{type = XType}, Fun, Serial0, Args) -> - Serial = fun (Bool) -> - case Serial0 of - _ when is_atom(Serial0) -> Serial0; - _ -> Serial0(Bool) - end +report_missing_decorators(Xs) -> + Mods = lists:usort(lists:append([rabbit_exchange_decorator:select(raw, D) || + #exchange{decorators = D} <- Xs])), + case [M || M <- Mods, code:which(M) =:= non_existing] of + [] -> ok; + M -> rabbit_log:warning("Missing exchange decorators: ~p~n", [M]) + end. + +callback(X = #exchange{type = XType, + decorators = Decorators}, Fun, Serial0, Args) -> + Serial = if is_function(Serial0) -> Serial0; + is_atom(Serial0) -> fun (_Bool) -> Serial0 end end, - [ok = apply(M, Fun, [Serial(M:serialise_events(X)) | Args]) - || M <- decorators()], + [ok = apply(M, Fun, [Serial(M:serialise_events(X)) | Args]) || + M <- rabbit_exchange_decorator:select(all, Decorators)], Module = type_to_module(XType), apply(Module, Fun, [Serial(Module:serialise_events()) | Args]). -policy_changed(X1, X2) -> callback(X1, policy_changed, none, [X1, X2]). +policy_changed(X = #exchange{type = XType, + decorators = Decorators}, + X1 = #exchange{decorators = Decorators1}) -> + D = rabbit_exchange_decorator:select(all, Decorators), + D1 = rabbit_exchange_decorator:select(all, Decorators1), + DAll = lists:usort(D ++ D1), + [ok = M:policy_changed(X, X1) || M <- [type_to_module(XType) | DAll]], + ok. -serialise_events(X = #exchange{type = Type}) -> - case [Serialise || M <- decorators(), - Serialise <- [M:serialise_events(X)], - Serialise == true] of - [] -> (type_to_module(Type)):serialise_events(); - _ -> true - end. +serialise_events(X = #exchange{type = Type, decorators = Decorators}) -> + lists:any(fun (M) -> M:serialise_events(X) end, + rabbit_exchange_decorator:select(all, Decorators)) + orelse (type_to_module(Type)):serialise_events(). serial(#exchange{name = XName} = X) -> Serial = case serialise_events(X) of @@ -144,9 +158,6 @@ serial(#exchange{name = XName} = X) -> (false) -> none end. -decorators() -> - [M || {_, M} <- rabbit_registry:lookup_all(exchange_decorator)]. - declare(XName, Type, Durable, AutoDelete, Internal, Args) -> X = rabbit_policy:set(#exchange{name = XName, type = Type, @@ -267,7 +278,8 @@ update_scratch(Name, App, Fun) -> Scratches2 = orddict:store( App, Fun(Scratch), Scratches1), X#exchange{scratches = Scratches2} - end) + end), + ok end). update(Name, Fun) -> @@ -278,9 +290,10 @@ update(Name, Fun) -> case Durable of true -> ok = mnesia:write(rabbit_durable_exchange, X1, write); _ -> ok - end; + end, + X1; [] -> - ok + not_found end. info_keys() -> ?INFO_KEYS. @@ -312,58 +325,67 @@ info_all(VHostPath) -> map(VHostPath, fun (X) -> info(X) end). info_all(VHostPath, Items) -> map(VHostPath, fun (X) -> info(X, Items) end). -%% Optimisation -route(#exchange{name = #resource{name = <<"">>, virtual_host = VHost}}, - #delivery{message = #basic_message{routing_keys = RKs}}) -> - [rabbit_misc:r(VHost, queue, RK) || RK <- lists:usort(RKs)]; - -route(X = #exchange{name = XName}, Delivery) -> - route1(Delivery, {queue:from_list([X]), XName, []}). - -route1(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))), - route1(Delivery, - lists:foldl(fun process_route/2, {WorkList1, SeenXs, QNames}, - DstNames)) +route(#exchange{name = #resource{virtual_host = VHost, name = RName} = XName, + decorators = Decorators} = X, + #delivery{message = #basic_message{routing_keys = RKs}} = Delivery) -> + case {RName, rabbit_exchange_decorator:select(route, Decorators)} of + {<<"">>, []} -> + %% Optimisation + [rabbit_misc:r(VHost, queue, RK) || RK <- lists:usort(RKs)]; + {_, SelectedDecorators} -> + lists:usort(route1(Delivery, SelectedDecorators, {[X], XName, []})) end. -process_alternate(#exchange{arguments = []}, Results) -> %% optimisation - Results; +route1(_, _, {[], _, QNames}) -> + QNames; +route1(Delivery, Decorators, + {[X = #exchange{type = Type} | WorkList], SeenXs, QNames}) -> + ExchangeDests = (type_to_module(Type)):route(X, Delivery), + DecorateDests = process_decorators(X, Decorators, Delivery), + AlternateDests = process_alternate(X, ExchangeDests), + route1(Delivery, Decorators, + lists:foldl(fun process_route/2, {WorkList, SeenXs, QNames}, + AlternateDests ++ DecorateDests ++ ExchangeDests)). + +process_alternate(#exchange{arguments = []}, _Results) -> %% optimisation + []; 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_alternate(_X, _Results) -> + []. + +process_decorators(_, [], _) -> %% optimisation + []; +process_decorators(X, Decorators, Delivery) -> + lists:append([Decorator:route(X, Delivery) || Decorator <- Decorators]). process_route(#resource{kind = exchange} = XName, {_WorkList, XName, _QNames} = Acc) -> Acc; process_route(#resource{kind = exchange} = XName, {WorkList, #resource{kind = exchange} = SeenX, QNames}) -> - {case lookup(XName) of - {ok, X} -> queue:in(X, WorkList); - {error, not_found} -> WorkList - end, gb_sets:from_list([SeenX, XName]), QNames}; + {cons_if_present(XName, WorkList), + gb_sets:from_list([SeenX, XName]), QNames}; process_route(#resource{kind = exchange} = XName, {WorkList, SeenXs, QNames} = Acc) -> case gb_sets:is_element(XName, SeenXs) of true -> Acc; - false -> {case lookup(XName) of - {ok, X} -> queue:in(X, WorkList); - {error, not_found} -> WorkList - end, gb_sets:add_element(XName, SeenXs), QNames} + false -> {cons_if_present(XName, WorkList), + gb_sets:add_element(XName, SeenXs), QNames} end; process_route(#resource{kind = queue} = QName, {WorkList, SeenXs, QNames}) -> {WorkList, SeenXs, [QName | QNames]}. +cons_if_present(XName, L) -> + case lookup(XName) of + {ok, X} -> [X | L]; + {error, not_found} -> L + end. + call_with_exchange(XName, Fun) -> rabbit_misc:execute_mnesia_tx_with_tail( fun () -> case mnesia:read({rabbit_exchange, XName}) of @@ -390,6 +412,10 @@ delete(XName, IfUnused) -> end end). +validate_binding(X = #exchange{type = XType}, Binding) -> + Module = type_to_module(XType), + Module:validate_binding(X, Binding). + maybe_auto_delete(#exchange{auto_delete = false}) -> not_deleted; maybe_auto_delete(#exchange{auto_delete = true} = X) -> @@ -431,8 +457,7 @@ peek_serial(XName, LockType) -> end. invalid_module(T) -> - rabbit_log:warning( - "Could not find exchange type ~s.~n", [T]), + rabbit_log:warning("Could not find exchange type ~s.~n", [T]), put({xtype_to_module, T}, rabbit_exchange_type_invalid), rabbit_exchange_type_invalid. |