%% 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-2022 VMware, Inc. or its affiliates. All rights reserved. %% -module(amqp10_client). -include("amqp10_client.hrl"). -include_lib("amqp10_common/include/amqp10_framing.hrl"). -export([open_connection/1, open_connection/2, close_connection/1, begin_session/1, begin_session_sync/1, begin_session_sync/2, end_session/1, attach_sender_link/3, attach_sender_link/4, attach_sender_link/5, attach_sender_link_sync/3, attach_sender_link_sync/4, attach_sender_link_sync/5, attach_receiver_link/3, attach_receiver_link/4, attach_receiver_link/5, attach_receiver_link/6, attach_receiver_link/7, attach_link/2, detach_link/1, send_msg/2, accept_msg/2, flow_link_credit/3, flow_link_credit/4, echo/1, link_handle/1, get_msg/1, get_msg/2, parse_uri/1 ]). -define(DEFAULT_TIMEOUT, 5000). -type snd_settle_mode() :: amqp10_client_session:snd_settle_mode(). -type rcv_settle_mode() :: amqp10_client_session:rcv_settle_mode(). -type terminus_durability() :: amqp10_client_session:terminus_durability(). -type target_def() :: amqp10_client_session:target_def(). -type source_def() :: amqp10_client_session:source_def(). -type attach_role() :: amqp10_client_session:attach_role(). -type attach_args() :: amqp10_client_session:attach_args(). -type filter() :: amqp10_client_session:filter(). -type properties() :: amqp10_client_session:properties(). -type connection_config() :: amqp10_client_connection:connection_config(). -opaque link_ref() :: #link_ref{}. -export_type([ link_ref/0, snd_settle_mode/0, rcv_settle_mode/0, terminus_durability/0, target_def/0, source_def/0, attach_role/0, attach_args/0, connection_config/0 ]). -ifdef (OTP_RELEASE). -if(?OTP_RELEASE >= 23). -compile({nowarn_deprecated_function, [{http_uri, decode, 1}]}). -endif. -endif. %% @doc Convenience function for opening a connection providing only an %% address and port. This uses anonymous sasl authentication. %% This is asynchronous and will notify success/closure to the caller using %% an amqp10_event of the following format: %% {amqp10_event, {connection, ConnectionPid, opened | {closed, Why}}} -spec open_connection(inet:socket_address() | inet:hostname(), inet:port_number()) -> supervisor:startchild_ret(). open_connection(Addr, Port) -> open_connection(#{address => Addr, port => Port, notify => self(), sasl => anon}). %% @doc Opens a connection using a connection_config map %% This is asynchronous and will notify success/closure to the caller using %% an amqp10_event of the following format: %% {amqp10_event, {connection, ConnectionPid, opened | {closed, Why}}} -spec open_connection(connection_config()) -> supervisor:startchild_ret(). open_connection(ConnectionConfig0) -> Notify = maps:get(notify, ConnectionConfig0, self()), NotifyWhenOpened = maps:get(notify_when_opened, ConnectionConfig0, self()), NotifyWhenClosed = maps:get(notify_when_closed, ConnectionConfig0, self()), amqp10_client_connection:open(ConnectionConfig0#{ notify => Notify, notify_when_opened => NotifyWhenOpened, notify_when_closed => NotifyWhenClosed }). %% @doc Opens a connection using a connection_config map %% This is asynchronous and will notify completion to the caller using %% an amqp10_event of the following format: %% {amqp10_event, {connection, ConnectionPid, {closed, Why}}} -spec close_connection(pid()) -> ok. close_connection(Pid) -> amqp10_client_connection:close(Pid, none). %% @doc Begins an amqp10 session using 'Connection'. %% This is asynchronous and will notify success/closure to the caller using %% an amqp10_event of the following format: %% {amqp10_event, {session, SessionPid, begun | {ended, Why}}} -spec begin_session(pid()) -> supervisor:startchild_ret(). begin_session(Connection) when is_pid(Connection) -> amqp10_client_connection:begin_session(Connection). %% @doc Synchronously begins an amqp10 session using 'Connection'. %% This is a convenience function that awaits the 'begun' event %% for the newly created session before returning. -spec begin_session_sync(pid()) -> supervisor:startchild_ret() | session_timeout. begin_session_sync(Connection) when is_pid(Connection) -> begin_session_sync(Connection, ?DEFAULT_TIMEOUT). %% @doc Synchronously begins an amqp10 session using 'Connection'. %% This is a convenience function that awaits the 'begun' event %% for the newly created session before returning. -spec begin_session_sync(pid(), non_neg_integer()) -> supervisor:startchild_ret() | session_timeout. begin_session_sync(Connection, Timeout) when is_pid(Connection) -> case begin_session(Connection) of {ok, Session} -> receive {amqp10_event, {session, Session, begun}} -> {ok, Session}; {amqp10_event, {session, Session, {ended, Err}}} -> {error, Err} after Timeout -> session_timeout end; Ret -> Ret end. %% @doc End an amqp10 session. %% This is asynchronous and will notify completion of the end request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {session, SessionPid, {ended, Why}}} -spec end_session(pid()) -> ok. end_session(Pid) -> amqp10_client_session:'end'(Pid). %% @doc Synchronously attach a link on 'Session'. %% This is a convenience function that awaits attached event %% for the link before returning. attach_sender_link_sync(Session, Name, Target) -> attach_sender_link_sync(Session, Name, Target, mixed). %% @doc Synchronously attach a link on 'Session'. %% This is a convenience function that awaits attached event %% for the link before returning. -spec attach_sender_link_sync(pid(), binary(), binary(), snd_settle_mode()) -> {ok, link_ref()} | link_timeout. attach_sender_link_sync(Session, Name, Target, SettleMode) -> attach_sender_link_sync(Session, Name, Target, SettleMode, none). %% @doc Synchronously attach a link on 'Session'. %% This is a convenience function that awaits attached event %% for the link before returning. -spec attach_sender_link_sync(pid(), binary(), binary(), snd_settle_mode(), terminus_durability()) -> {ok, link_ref()} | link_timeout. attach_sender_link_sync(Session, Name, Target, SettleMode, Durability) -> {ok, Ref} = attach_sender_link(Session, Name, Target, SettleMode, Durability), receive {amqp10_event, {link, Ref, attached}} -> {ok, Ref}; {amqp10_event, {link, Ref, {detached, Err}}} -> {error, Err} after ?DEFAULT_TIMEOUT -> link_timeout end. %% @doc Attaches a sender link to a target. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}} -spec attach_sender_link(pid(), binary(), binary()) -> {ok, link_ref()}. attach_sender_link(Session, Name, Target) -> % mixed should work with any type of msg attach_sender_link(Session, Name, Target, mixed). %% @doc Attaches a sender link to a target. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}} -spec attach_sender_link(pid(), binary(), binary(), snd_settle_mode()) -> {ok, link_ref()}. attach_sender_link(Session, Name, Target, SettleMode) -> attach_sender_link(Session, Name, Target, SettleMode, none). %% @doc Attaches a sender link to a target. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}} -spec attach_sender_link(pid(), binary(), binary(), snd_settle_mode(), terminus_durability()) -> {ok, link_ref()}. attach_sender_link(Session, Name, Target, SettleMode, Durability) -> AttachArgs = #{name => Name, role => {sender, #{address => Target, durable => Durability}}, snd_settle_mode => SettleMode, rcv_settle_mode => first}, amqp10_client_session:attach(Session, AttachArgs). %% @doc Attaches a receiver link to a source. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}} -spec attach_receiver_link(pid(), binary(), binary()) -> {ok, link_ref()}. attach_receiver_link(Session, Name, Source) -> attach_receiver_link(Session, Name, Source, settled). %% @doc Attaches a receiver link to a source. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}} -spec attach_receiver_link(pid(), binary(), binary(), snd_settle_mode()) -> {ok, link_ref()}. attach_receiver_link(Session, Name, Source, SettleMode) -> attach_receiver_link(Session, Name, Source, SettleMode, none). %% @doc Attaches a receiver link to a source. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}} -spec attach_receiver_link(pid(), binary(), binary(), snd_settle_mode(), terminus_durability()) -> {ok, link_ref()}. attach_receiver_link(Session, Name, Source, SettleMode, Durability) -> attach_receiver_link(Session, Name, Source, SettleMode, Durability, #{}). %% @doc Attaches a receiver link to a source. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}} -spec attach_receiver_link(pid(), binary(), binary(), snd_settle_mode(), terminus_durability(), filter()) -> {ok, link_ref()}. attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter) -> attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter, #{}). %% @doc Attaches a receiver link to a source. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, attached | {detached, Why}}} -spec attach_receiver_link(pid(), binary(), binary(), snd_settle_mode(), terminus_durability(), filter(), properties()) -> {ok, link_ref()}. attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter, Properties) -> AttachArgs = #{name => Name, role => {receiver, #{address => Source, durable => Durability}, self()}, snd_settle_mode => SettleMode, rcv_settle_mode => first, filter => Filter, properties => Properties}, amqp10_client_session:attach(Session, AttachArgs). -spec attach_link(pid(), attach_args()) -> {ok, link_ref()}. attach_link(Session, AttachArgs) -> amqp10_client_session:attach(Session, AttachArgs). %% @doc Detaches a link. %% This is asynchronous and will notify completion of the attach request to the %% caller using an amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, {detached, Why}}} -spec detach_link(link_ref()) -> _. detach_link(#link_ref{link_handle = Handle, session = Session}) -> amqp10_client_session:detach(Session, Handle). %% @doc Grant credit to a sender. %% The amqp10_client will automatically grant more credit to the sender when %% the remaining link credit falls below the value of RenewWhenBelow. %% If RenewWhenBelow is 'never' the client will never grant new credit. Instead %% the caller will be notified when the link_credit reaches 0 with an %% amqp10_event of the following format: %% {amqp10_event, {link, LinkRef, credit_exhausted}} -spec flow_link_credit(link_ref(), Credit :: non_neg_integer(), RenewWhenBelow :: never | non_neg_integer()) -> ok. flow_link_credit(Ref, Credit, RenewWhenBelow) -> flow_link_credit(Ref, Credit, RenewWhenBelow, false). -spec flow_link_credit(link_ref(), Credit :: non_neg_integer(), RenewWhenBelow :: never | non_neg_integer(), Drain :: boolean()) -> ok. flow_link_credit(#link_ref{role = receiver, session = Session, link_handle = Handle}, Credit, RenewWhenBelow, Drain) -> Flow = #'v1_0.flow'{link_credit = {uint, Credit}, drain = Drain}, ok = amqp10_client_session:flow(Session, Handle, Flow, RenewWhenBelow). %% @doc Request that the sender's flow state is echoed back %% This may be used to determine when the Link has finally quiesced. %% see ยง2.6.10 of the spec echo(#link_ref{role = receiver, session = Session, link_handle = Handle}) -> Flow = #'v1_0.flow'{link_credit = {uint, 0}, echo = true}, ok = amqp10_client_session:flow(Session, Handle, Flow, 0). %%% messages %% @doc Send a message on a the link referred to be the 'LinkRef'. %% Returns ok for "async" transfers when messages are sent with settled=true %% else it returns the delivery state from the disposition -spec send_msg(link_ref(), amqp10_msg:amqp10_msg()) -> ok | {error, insufficient_credit | link_not_found | half_attached}. send_msg(#link_ref{role = sender, session = Session, link_handle = Handle}, Msg0) -> Msg = amqp10_msg:set_handle(Handle, Msg0), amqp10_client_session:transfer(Session, Msg, ?DEFAULT_TIMEOUT). %% @doc Accept a message on a the link referred to be the 'LinkRef'. -spec accept_msg(link_ref(), amqp10_msg:amqp10_msg()) -> ok. accept_msg(#link_ref{role = receiver, session = Session}, Msg) -> DeliveryId = amqp10_msg:delivery_id(Msg), amqp10_client_session:disposition(Session, receiver, DeliveryId, DeliveryId, true, accepted). %% @doc Get a single message from a link. %% Flows a single link credit then awaits delivery or timeout. -spec get_msg(link_ref()) -> {ok, amqp10_msg:amqp10_msg()} | {error, timeout}. get_msg(LinkRef) -> get_msg(LinkRef, ?DEFAULT_TIMEOUT). %% @doc Get a single message from a link. %% Flows a single link credit then awaits delivery or timeout. -spec get_msg(link_ref(), non_neg_integer()) -> {ok, amqp10_msg:amqp10_msg()} | {error, timeout}. get_msg(#link_ref{role = receiver} = Ref, Timeout) -> %flow 1 ok = flow_link_credit(Ref, 1, never), % wait for transfer receive {amqp10_msg, Ref, Message} -> {ok, Message} after Timeout -> {error, timeout} end. %% @doc Get the link handle from a LinkRef -spec link_handle(link_ref()) -> non_neg_integer(). link_handle(#link_ref{link_handle = Handle}) -> Handle. -spec parse_uri(string()) -> {ok, connection_config()} | {error, term()}. parse_uri(Uri) -> case uri_string:parse(Uri) of Map when is_map(Map) -> try {ok, parse_result(Map)} catch throw:Err -> {error, Err} end; {error, _, _} = Err -> Err end. parse_result(Map) -> _ = case maps:get(path, Map, "/") of "/" -> ok; "" -> ok; _ -> throw(path_segment_not_supported) end, Scheme = maps:get(scheme, Map, "amqp"), UserInfo = maps:get(userinfo, Map, undefined), Host = maps:get(host, Map), DefaultPort = case Scheme of "amqp" -> 5672; "amqps" -> 5671 end, Port = maps:get(port, Map, DefaultPort), Query0 = maps:get(query, Map, ""), Query = maps:from_list(uri_string:dissect_query(Query0)), Sasl = case Query of #{"sasl" := "anon"} -> anon; #{"sasl" := "plain"} when UserInfo =:= undefined orelse length(UserInfo) =:= 0 -> throw(plain_sasl_missing_userinfo); _ -> case UserInfo of [] -> none; undefined -> none; U -> parse_usertoken(U) end end, Ret0 = maps:fold(fun("idle_time_out", V, Acc) -> Acc#{idle_time_out => list_to_integer(V)}; ("max_frame_size", V, Acc) -> Acc#{max_frame_size => list_to_integer(V)}; ("hostname", V, Acc) -> Acc#{hostname => list_to_binary(V)}; ("container_id", V, Acc) -> Acc#{container_id => list_to_binary(V)}; ("transfer_limit_margin", V, Acc) -> Acc#{transfer_limit_margin => list_to_integer(V)}; (_, _, Acc) -> Acc end, #{address => Host, hostname => to_binary(Host), port => Port, sasl => Sasl}, Query), case Scheme of "amqp" -> Ret0; "amqps" -> TlsOpts = parse_tls_opts(Query), Ret0#{tls_opts => {secure_port, TlsOpts}} end. parse_usertoken(undefined) -> none; parse_usertoken("") -> none; parse_usertoken(U) -> [User, Pass] = string:tokens(U, ":"), {plain, to_binary(uri_string:percent_decode(User)), to_binary(uri_string:percent_decode(Pass))}. parse_tls_opts(M) -> lists:sort(maps:fold(fun parse_tls_opt/3, [], M)). parse_tls_opt(K, V, Acc) when K =:= "cacertfile" orelse K =:= "certfile" orelse K =:= "keyfile" -> [{to_atom(K), V} | Acc]; parse_tls_opt("verify", V, Acc) when V =:= "verify_none" orelse V =:= "verify_peer" -> [{verify, to_atom(V)} | Acc]; parse_tls_opt("verify", _V, _Acc) -> throw({invalid_option, verify}); parse_tls_opt("versions", V, Acc) -> Parts = string:tokens(V, ","), Versions = [try_to_existing_atom(P) || P <- Parts], [{versions, Versions} | Acc]; parse_tls_opt("fail_if_no_peer_cert", V, Acc) when V =:= "true" orelse V =:= "false" -> [{fail_if_no_peer_cert, to_atom(V)} | Acc]; parse_tls_opt("fail_if_no_peer_cert", _V, _Acc) -> throw({invalid_option, fail_if_no_peer_cert}); parse_tls_opt("server_name_indication", "disable", Acc) -> [{server_name_indication, disable} | Acc]; parse_tls_opt("server_name_indication", V, Acc) -> [{server_name_indication, V} | Acc]; parse_tls_opt(_K, _V, Acc) -> Acc. to_atom(X) when is_list(X) -> list_to_atom(X). to_binary(X) when is_list(X) -> list_to_binary(X). try_to_existing_atom(L) when is_list(L) -> try list_to_existing_atom(L) of A -> A catch _:badarg -> throw({non_existent_atom, L}) end. -ifdef(TEST). -include_lib("eunit/include/eunit.hrl"). parse_uri_test_() -> [?_assertEqual({ok, #{address => "my_host", port => 9876, hostname => <<"my_host">>, sasl => none}}, parse_uri("amqp://my_host:9876")), %% port defaults ?_assertMatch({ok, #{port := 5671}}, parse_uri("amqps://my_host")), ?_assertMatch({ok, #{port := 5672}}, parse_uri("amqp://my_host")), ?_assertEqual({ok, #{address => "my_proxy", port => 9876, hostname => <<"my_host">>, container_id => <<"my_container">>, idle_time_out => 60000, max_frame_size => 512, tls_opts => {secure_port, []}, sasl => {plain, <<"fred">>, <<"passw">>}}}, parse_uri("amqps://fred:passw@my_proxy:9876?sasl=plain&" "hostname=my_host&container_id=my_container&" "max_frame_size=512&idle_time_out=60000")), %% ensure URI encoded usernames and passwords are decodeded ?_assertEqual({ok, #{address => "my_proxy", port => 9876, hostname => <<"my_proxy">>, sasl => {plain, <<"fr/ed">>, <<"pa/ssw">>}}}, parse_uri("amqp://fr%2Fed:pa%2Fssw@my_proxy:9876")), %% make sasl plain implicit when username and password is present ?_assertEqual({ok, #{address => "my_proxy", port => 9876, hostname => <<"my_proxy">>, sasl => {plain, <<"fred">>, <<"passw">>}}}, parse_uri("amqp://fred:passw@my_proxy:9876")), ?_assertEqual( {ok, #{address => "my_proxy", port => 9876, hostname => <<"my_proxy">>, tls_opts => {secure_port, [{cacertfile, "/etc/cacertfile.pem"}, {certfile, "/etc/certfile.pem"}, {fail_if_no_peer_cert, true}, {keyfile, "/etc/keyfile.key"}, {server_name_indication, "frazzle"}, {verify, verify_none}, {versions, ['tlsv1.1', 'tlsv1.2']} ]}, sasl => {plain, <<"fred">>, <<"passw">>}}}, parse_uri("amqps://fred:passw@my_proxy:9876?sasl=plain" "&cacertfile=/etc/cacertfile.pem&certfile=/etc/certfile.pem" "&keyfile=/etc/keyfile.key&verify=verify_none" "&fail_if_no_peer_cert=true" "&server_name_indication=frazzle" "&versions=tlsv1.1,tlsv1.2")), %% invalid tls version ?_assertEqual({error, {non_existent_atom, "tlsv1.9999999"}}, parse_uri("amqps://fred:passw@my_proxy:9876?sasl=plain&" ++ "versions=tlsv1.1,tlsv1.9999999")), ?_assertEqual( {ok, #{address => "my_proxy", port => 9876, hostname => <<"my_proxy">>, tls_opts => {secure_port, [{server_name_indication, disable}]}, sasl => {plain, <<"fred">>, <<"passw">>}}}, parse_uri("amqps://fred:passw@my_proxy:9876?sasl=plain" "&server_name_indication=disable")), ?_assertEqual({error, {invalid_option, verify}}, parse_uri("amqps://fred:passw@my_proxy:9876?sasl=plain&" ++ "cacertfile=/etc/cacertfile.pem&certfile=/etc/certfile.pem&" ++ "keyfile=/etc/keyfile.key&verify=verify_bananas")), ?_assertEqual({error, {invalid_option, fail_if_no_peer_cert}}, parse_uri("amqps://fred:passw@my_proxy:9876?sasl=plain&" ++ "cacertfile=/etc/cacertfile.pem&certfile=/etc/certfile.pem&" ++ "keyfile=/etc/keyfile.key&fail_if_no_peer_cert=banana")), ?_assertEqual({error, plain_sasl_missing_userinfo}, parse_uri("amqp://my_host:9876?sasl=plain")), ?_assertEqual({error, path_segment_not_supported}, parse_uri("amqp://my_host/my_path_segment:9876")) ]. -endif.