Add capabilities to attach function

for sender and receiver links

Author: MarcialRosales

deps/amqp10_client/src/amqp10_client.erl

@@ -20,14 +20,17 @@
          attach_sender_link/3,
          attach_sender_link/4,
          attach_sender_link/5,
+         attach_sender_link/6,
          attach_sender_link_sync/3,
          attach_sender_link_sync/4,
          attach_sender_link_sync/5,
+         attach_sender_link_sync/6,
          attach_receiver_link/3,
          attach_receiver_link/4,
          attach_receiver_link/5,
          attach_receiver_link/6,
          attach_receiver_link/7,
+         attach_receiver_link/8,
          attach_link/2,
          detach_link/1,
          send_msg/2,
@@ -182,8 +185,19 @@ attach_sender_link_sync(Session, Name, Target, SettleMode) ->
                               snd_settle_mode(), terminus_durability()) ->
     {ok, link_ref()} | link_timeout.
 attach_sender_link_sync(Session, Name, Target, SettleMode, Durability) ->
+    attach_sender_link_sync(Session, Name, Target, SettleMode, Durability, 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(),
+                              binary() | list()) ->
+    {ok, link_ref()} | link_timeout.
+
+attach_sender_link_sync(Session, Name, Target, SettleMode, Durability, Capabilities) ->
     {ok, Ref} = attach_sender_link(Session, Name, Target, SettleMode,
-                                   Durability),
+                                   Durability, Capabilities),
     receive
         {amqp10_event, {link, Ref, attached}} ->
             {ok, Ref};
@@ -226,6 +240,23 @@ attach_sender_link(Session, Name, Target, SettleMode, Durability) ->
                    rcv_settle_mode => first},
     amqp10_client_session:attach(Session, AttachArgs).
 
+%% @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(),
+                         binary() | list()) ->
+    {ok, link_ref()}.
+attach_sender_link(Session, Name, Target, SettleMode, Durability, Capabilities) ->
+    AttachArgs = #{name => Name,
+                   role => {sender, #{address => Target,
+                                      durable => Durability,
+                                      capabilities => Capabilities}},
+                   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:
@@ -272,7 +303,19 @@ attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter) ->
 -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)
+attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter, Properties) ->
+    attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter, Properties, 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(), filter(), properties(),
+                           binary() | list()) ->
+    {ok, link_ref()}.
+attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter, 
+    Properties, Capabilities)
   when is_pid(Session) andalso
        is_binary(Name) andalso
        is_binary(Source) andalso
@@ -281,10 +324,13 @@ attach_receiver_link(Session, Name, Source, SettleMode, Durability, Filter, Prop
         SettleMode == mixed) andalso
        is_atom(Durability) andalso
        is_map(Filter) andalso
-       is_map(Properties) ->
+       is_map(Properties) andalso 
+       is_list(Capabilities) orelse 
+       (Capabilities == none orelse is_binary(Capabilities)) ->
     AttachArgs = #{name => Name,
                    role => {receiver, #{address => Source,
-                                        durable => Durability}, self()},
+                                        durable => Durability,
+                                        capabilities => Capabilities}, self()},
                    snd_settle_mode => SettleMode,
                    rcv_settle_mode => first,
                    filter => Filter,

deps/amqp10_client/src/amqp10_client_session.erl

@@ -713,20 +713,24 @@ make_source(#{role := {sender, _}}) ->
 make_source(#{role := {receiver, #{address := Address} = Source, _Pid}, filter := Filter}) ->
     Durable = translate_terminus_durability(maps:get(durable, Source, none)),
     TranslatedFilter = translate_filters(Filter),
+    Capabilities = translate_terminus_capabilities(maps:get(capabilities, Source, none)),
     #'v1_0.source'{address = {utf8, Address},
                    durable = {uint, Durable},
-                   filter = TranslatedFilter}.
+                   filter = TranslatedFilter,
+                   capabilities = Capabilities}.
 
 make_target(#{role := {receiver, _Source, _Pid}}) ->
     #'v1_0.target'{};
 make_target(#{role := {sender, #{address := Address} = Target}}) ->
     Durable = translate_terminus_durability(maps:get(durable, Target, none)),
+    Capabilities = translate_terminus_capabilities(maps:get(capabilities, Target, none)),
     TargetAddr = case is_binary(Address) of
                      true -> {utf8, Address};
                      false -> Address
                  end,
     #'v1_0.target'{address = TargetAddr,
-                   durable = {uint, Durable}}.
+                   durable = {uint, Durable},
+                   capabilities = Capabilities}.
 
 max_message_size(#{max_message_size := Size})
   when is_integer(Size) andalso
@@ -771,6 +775,19 @@ filter_value_type({T, _} = V) when is_atom(T) ->
     %% looks like an already tagged type, just pass it through
     V.
 
+translate_terminus_capabilities(none) ->
+    undefined;
+translate_terminus_capabilities(Capabilities) when is_binary(Capabilities) ->
+    {utf8, Capabilities};
+translate_terminus_capabilities(CapabilitiesList) when is_list(CapabilitiesList) ->
+    {list, [filter_capability(V) || V <- CapabilitiesList]}.
+
+filter_capability(V) when is_binary(V) ->
+    {utf8, V};
+filter_capability({T, _} = V) when is_atom(T) ->
+    %% looks like an already tagged type, just pass it through
+    V.
+        
 % https://people.apache.org/~rgodfrey/amqp-1.0/apache-filters.html
 translate_legacy_amqp_headers_binding(LegacyHeaders) ->
     {map,