Packages
amqp_client
4.2.0
4.2.1
4.2.0
retired
4.1.6
4.1.5
retired
4.1.5-rc.2
retired
4.1.5-rc.1
retired
4.1.5-1
4.0.3
4.0.3-rc.1
3.13.0-rc.2
3.13.0-rc.1
3.12.14
3.12.13
3.12.12
3.12.11
3.12.10
3.12.9
3.12.8
3.12.7
3.12.6
3.12.5
3.12.4
3.12.3
3.12.2
3.12.1
3.12.0
3.12.0-rc.4
3.11.28
3.11.27
3.11.26
3.11.25
3.11.24
3.11.23
3.11.22
3.11.21
3.11.20
3.11.19
3.11.18
3.11.17
3.11.13
3.11.12
3.11.11
3.11.10
3.11.9
3.11.8
3.11.7
3.11.6
3.11.5
3.11.4
3.11.3
3.11.2
3.11.1
3.11.0
3.11.0-rc.2
3.11.0-rc.1
3.11.0-1
3.10.20
3.10.19
3.10.18
3.10.17
3.10.16
3.10.15
3.10.14
3.10.13
3.10.12
3.10.11
3.10.10
3.10.9
3.10.8
3.10.7
3.10.6
3.10.5
3.10.4
3.10.3
3.10.2
3.10.1
3.10.0
3.10.0-rc.6
3.10.0-rc.5
3.9.29
3.9.28
3.9.27
3.9.26
3.9.25
3.9.24
3.9.23
3.9.22
3.9.21
3.9.20
3.9.19
3.9.18
3.9.17
3.9.16
3.9.15
3.9.11
3.9.8
3.9.5
3.9.4
3.9.1
3.8.35
3.8.34
3.8.33
3.8.32
3.8.31
3.8.30
3.8.25
3.8.21
3.8.19
3.8.14
3.8.11
3.8.10
3.8.10-rc.1
3.8.9
3.8.7
3.8.6
3.8.6-rc.2
3.8.6-rc.1
3.8.5
3.8.5-rc.2
3.8.5-rc.1
3.8.4
3.8.4-rc.3
3.8.4-rc.1
3.8.3
3.8.3-rc.2
3.8.3-rc.1
3.8.2
3.8.2-rc.1
3.8.1
3.8.1-rc.1
3.8.0
3.8.0-rc.3
3.8.0-rc.2
3.8.0-rc.1
3.7.28
3.7.27
3.7.27-rc.2
3.7.27-rc.1
3.7.26
3.7.25
3.7.25-rc.1
3.7.24
3.7.24-rc.2
3.7.24-rc.1
3.7.23
3.7.23-rc.1
3.7.22
3.7.22-rc.2
3.7.22-rc.1
3.7.21
3.7.20
3.7.20-rc.2
3.7.20-rc.1
3.7.19
3.7.18
3.7.18-rc.1
3.7.17
3.7.17-rc.3
3.7.17-rc.2
3.7.15
3.7.14
3.7.14-rc.2
3.7.14-rc.1
3.7.13
3.7.13-rc.2
3.7.13-rc.1
3.7.12
3.7.12-rc.2
3.7.12-rc.1
3.7.11
3.7.11-rc.2
3.7.11-rc.1
3.7.10-rc.4
3.7.10-rc.3
3.7.10-rc.2
3.7.10-rc.1
3.7.9
3.7.9-rc.3
3.7.9-rc.2
3.7.8
3.7.8-rc.4
3.7.8-rc.3
3.7.8-rc.2
3.7.8-rc.1
3.7.7
3.7.7-rc.2
3.7.7-rc.1
3.7.6
3.7.6-rc.2
3.7.6-rc.1
3.7.5
3.7.5-rc.1
3.7.4
3.7.4-rc.4
3.7.4-rc.3
3.7.4-rc.2
3.7.4-rc.1
3.7.3
3.7.3-rc.2
3.7.3-rc.1
3.7.2
3.7.1
3.7.0-rc.2
3.6.16
3.6.16-rc.1
3.6.15
3.6.15-rc.1
3.6.14
3.6.13
3.6.12
3.6.11
3.6.10
3.6.9
3.6.8
3.6.7
3.6.7-pre.1
3.5.6
3.5.0
3.4.0
3.3.5
3.0.2
0.0.0-rc.1
RabbitMQ AMQP Client
Retired package: Release invalid - invalid
Current section
Files
Jump to
Current section
Files
src/rabbit_routing_util.erl
%% This Source Code Form is subject to the terms of the Mozilla Public
%% License, v. 2.0. If a copy of the MPL was not distributed with this
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
%%
%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
%%
-module(rabbit_routing_util).
-export([init_state/0, dest_prefixes/0, all_dest_prefixes/0]).
-export([ensure_endpoint/4, ensure_endpoint/5, ensure_binding/3]).
-export([dest_temp_queue/1]).
-include("amqp_client.hrl").
-include("rabbit_routing_prefixes.hrl").
%%----------------------------------------------------------------------------
init_state() -> sets:new().
dest_prefixes() -> [?EXCHANGE_PREFIX, ?TOPIC_PREFIX, ?QUEUE_PREFIX,
?AMQQUEUE_PREFIX, ?REPLY_QUEUE_PREFIX].
all_dest_prefixes() -> [?TEMP_QUEUE_PREFIX | dest_prefixes()].
%% --------------------------------------------------------------------------
ensure_endpoint(Dir, Channel, Endpoint, State) ->
ensure_endpoint(Dir, Channel, Endpoint, [], State).
ensure_endpoint(source, Channel, {exchange, {Name, _}}, Params, State) ->
check_exchange(Name, Channel,
proplists:get_value(check_exchange, Params, false)),
Method = queue_declare_method(#'queue.declare'{}, exchange, Params),
#'queue.declare_ok'{queue = Queue} = amqp_channel:call(Channel, Method),
{ok, Queue, State};
ensure_endpoint(source, Channel, {topic, _}, Params, State) ->
Method = queue_declare_method(#'queue.declare'{}, topic, Params),
#'queue.declare_ok'{queue = Queue} = amqp_channel:call(Channel, Method),
{ok, Queue, State};
ensure_endpoint(_Dir, _Channel, {queue, undefined}, _Params, State) ->
{ok, undefined, State};
ensure_endpoint(_, Channel, {queue, Name}, Params, State) ->
Params1 = rabbit_misc:pmerge(durable, true, Params),
Queue = list_to_binary(Name),
State1 = case sets:is_element(Queue, State) of
true -> State;
_ -> Method = queue_declare_method(
#'queue.declare'{queue = Queue,
nowait = true},
queue, Params1),
case Method#'queue.declare'.nowait of
true -> amqp_channel:cast(Channel, Method);
false -> amqp_channel:call(Channel, Method)
end,
sets:add_element(Queue, State)
end,
{ok, Queue, State1};
ensure_endpoint(dest, Channel, {exchange, {Name, _}}, Params, State) ->
check_exchange(Name, Channel,
proplists:get_value(check_exchange, Params, false)),
{ok, undefined, State};
ensure_endpoint(dest, _Ch, {topic, _}, _Params, State) ->
{ok, undefined, State};
ensure_endpoint(_, _Ch, {amqqueue, Name}, _Params, State) ->
{ok, list_to_binary(Name), State};
ensure_endpoint(_, _Ch, {reply_queue, Name}, _Params, State) ->
{ok, list_to_binary(Name), State};
ensure_endpoint(_Direction, _Ch, _Endpoint, _Params, _State) ->
{error, invalid_endpoint}.
%% --------------------------------------------------------------------------
ensure_binding(QueueBin, {"", Queue}, _Channel) ->
%% i.e., we should only be asked to bind to the default exchange a
%% queue with its own name
QueueBin = list_to_binary(Queue),
ok;
ensure_binding(Queue, {Exchange, RoutingKey}, Channel) ->
#'queue.bind_ok'{} =
amqp_channel:call(Channel,
#'queue.bind'{
queue = Queue,
exchange = list_to_binary(Exchange),
routing_key = list_to_binary(RoutingKey)}),
ok.
%% --------------------------------------------------------------------------
dest_temp_queue({temp_queue, Name}) -> Name;
dest_temp_queue(_) -> none.
%% --------------------------------------------------------------------------
check_exchange(_, _, false) ->
ok;
check_exchange(ExchangeName, Channel, true) ->
XDecl = #'exchange.declare'{ exchange = list_to_binary(ExchangeName),
passive = true },
#'exchange.declare_ok'{} = amqp_channel:call(Channel, XDecl),
ok.
update_queue_declare_arguments(Method, Params) ->
Method#'queue.declare'{arguments =
proplists:get_value(arguments, Params, [])}.
update_queue_declare_exclusive(Method, Params) ->
case proplists:get_value(exclusive, Params) of
undefined -> Method;
Val -> Method#'queue.declare'{exclusive = Val}
end.
update_queue_declare_auto_delete(Method, Params) ->
case proplists:get_value(auto_delete, Params) of
undefined -> Method;
Val -> Method#'queue.declare'{auto_delete = Val}
end.
update_queue_declare_nowait(Method, Params) ->
case proplists:get_value(nowait, Params) of
undefined -> Method;
Val -> Method#'queue.declare'{nowait = Val}
end.
queue_declare_method(#'queue.declare'{} = Method, Type, Params) ->
%% defaults
Method1 = case proplists:get_value(durable, Params, false) of
true -> Method#'queue.declare'{durable = true};
false -> Method#'queue.declare'{auto_delete = true,
exclusive = true}
end,
%% set the rest of queue.declare fields from Params
Method2 = lists:foldl(fun (F, Acc) -> F(Acc, Params) end,
Method1, [fun update_queue_declare_arguments/2,
fun update_queue_declare_exclusive/2,
fun update_queue_declare_auto_delete/2,
fun update_queue_declare_nowait/2]),
case {Type, proplists:get_value(subscription_queue_name_gen, Params)} of
{topic, SQNG} when is_function(SQNG) ->
Method2#'queue.declare'{queue = SQNG()};
{exchange, SQNG} when is_function(SQNG) ->
Method2#'queue.declare'{queue = SQNG()};
_ ->
Method2
end.