Add capabilities parameter

Author: MarcialRosales

deps/amqp10_client/src/amqp10_client.erl

@@ -20,6 +20,7 @@
          attach_sender_link/3,
          attach_sender_link/4,
          attach_sender_link/5,
+         attach_sender_link_sync/2,
          attach_sender_link_sync/3,
          attach_sender_link_sync/4,
          attach_sender_link_sync/5,
@@ -197,6 +198,19 @@ attach_sender_link_sync(Session, Name, Target, SettleMode, Durability) ->
     after ?TIMEOUT -> link_timeout
     end.
 
+-spec attach_sender_link_sync(pid(), attach_args()) -> {ok, link_ref()} | link_timeout.
+attach_sender_link_sync(Session, AttachArgs) ->
+    {ok, Ref} =     amqp10_client_session:attach(Session, AttachArgs),
+    receive
+        {amqp10_event, {link, Ref, attached}} ->
+            {ok, Ref};
+        {amqp10_event, {link, Ref, {attached, #'v1_0.attach'{}}}} ->
+            {ok, Ref};
+        {amqp10_event, {link, Ref, {detached, Err}}} ->
+            {error, Err}
+    after ?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:

deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl

@@ -79,7 +79,7 @@ connect_source(State = #{name := Name,
                         on_publish -> unsettled;
                         on_confirm -> unsettled
                     end,
-    AttachFun = fun amqp10_client:attach_receiver_link/5,
+    AttachFun = fun amqp10_client:attach_link/2,    
     {Conn, Sess, LinkRef} = connect(Name, SndSettleMode, Uri, "receiver", Addr, Src,
                                     AttachFun),
     State#{source => Src#{current => #{conn => Conn,
@@ -97,7 +97,7 @@ connect_dest(State = #{name := Name,
                         on_publish -> settled;
                         on_confirm -> unsettled
                     end,
-    AttachFun = fun amqp10_client:attach_sender_link_sync/5,
+    AttachFun = fun amqp10_client:attach_sender_link_sync/2,
     {Conn, Sess, LinkRef} = connect(Name, SndSettleMode, Uri, "sender", Addr, Dst,
                                     AttachFun),
     %% wait for link credit here as if there are messages waiting we may try
@@ -126,9 +126,31 @@ connect(Name, SndSettleMode, Uri, Postfix, Addr, Map, AttachFun) ->
     % needs to be sync, i.e. awaits the 'attach' event as
     % else we may try to use the link before it is ready
     Durability = maps:get(durability, Map, unsettled_state),
-    {ok, LinkRef} = AttachFun(Sess, LinkName, Addr,
-                              SndSettleMode,
-                              Durability),
+    AttachArgs = case Postfix of 
+        "sender" -> 
+            SenderCapabilities = maps:get(sender_capabilities, Map, undefined),   
+            #{
+                name => LinkName,
+                role => {sender, #{address => Addr,
+                                    durable => Durability,
+                                    capabilities => SenderCapabilities}},
+                snd_settle_mode => SndSettleMode,
+                rcv_settle_mode => first
+            };
+        "receiver" ->
+            ReceiverCapabilities = maps:get(receiver_capabilities, Map, undefined),
+            #{
+                name => LinkName,
+                role => {receiver, #{address => Addr,
+                                    durable => Durability,
+                                    capabilities => ReceiverCapabilities}, self()},
+                snd_settle_mode => SndSettleMode,
+                rcv_settle_mode => first,
+                filter => #{},
+                properties => #{}
+            }   
+        end,
+    {ok, LinkRef} = AttachFun(Sess, AttachArgs),                                         
     {Conn, Sess, LinkRef}.
 
 -spec init_source(state()) -> state().

deps/rabbitmq_shovel/src/rabbit_shovel_parameters.erl

@@ -133,7 +133,8 @@ amqp10_src_validation(_Def, User) ->
      {<<"src-uri">>, validate_uri_fun(User), mandatory},
      {<<"src-address">>, fun rabbit_parameter_validation:binary/2, mandatory},
      {<<"src-prefetch-count">>, fun rabbit_parameter_validation:number/2, optional},
-     {<<"src-delete-after">>, fun validate_delete_after/2, optional}
+     {<<"src-delete-after">>, fun validate_delete_after/2, optional},
+     {<<"src-capabilities">>, fun validate_link_capabilities/2, optional}
     ].
 
 amqp091_src_validation(_Def, User) ->
@@ -168,7 +169,8 @@ amqp10_dest_validation(_Def, User) ->
      {<<"dest-application-properties">>, fun validate_amqp10_map/2, optional},
      {<<"dest-message-annotations">>, fun validate_amqp10_map/2, optional},
      % TODO: restrict to allowed fields
-     {<<"dest-properties">>, fun validate_amqp10_map/2, optional}
+     {<<"dest-properties">>, fun validate_amqp10_map/2, optional},
+     {<<"dest-capabilities">>, fun validate_link_capabilities/2, optional}
     ].
 
 amqp091_dest_validation(_Def, User) ->
@@ -226,6 +228,19 @@ validate_params_user(#amqp_params_direct{virtual_host = VHost},
 validate_params_user(#amqp_params_network{}, _User) ->
     ok.
 
+validate_link_capabilities(Name, ListOfSymbols) when is_list(ListOfSymbols) ->
+    case lists:all(fun(V) when is_binary(V) -> true;
+                      (_V) -> false
+                    end, ListOfSymbols) of 
+        true -> ok;
+        false -> {error, "~ts should have list of binaries, actually was ~tp", 
+            [Name, ListOfSymbols]}
+    end;
+validate_link_capabilities(_Name, Symbol) when is_binary(Symbol) ->
+    ok;
+validate_link_capabilities(Name, Symbol) ->
+    {error, "~ts should be a binary, actually was ~tp", [Name, Symbol]}.
+
 validate_delete_after(_Name, <<"never">>)          -> ok;
 validate_delete_after(_Name, <<"queue-length">>)   -> ok;
 validate_delete_after(_Name, N) when is_integer(N), N >= 0 -> ok;
@@ -314,11 +329,13 @@ parse_amqp10_dest({_VHost, _Name}, _ClusterName, Def, SourceHeaders) ->
     MessageAnns =
         rabbit_data_coercion:to_proplist(
             pget(<<"dest-message-annotations">>, Def, [])),
+    ReceiverCapabilities = pget(<<"dest-capabilities">>, Def),
     #{module => rabbit_amqp10_shovel,
       uris => Uris,
       target_address => Address,
       message_annotations => maps:from_list(MessageAnns),
       application_properties => maps:from_list(AppProperties ++ SourceHeaders),
+      receiver_capabilities => ReceiverCapabilities,
       properties => maps:from_list(
                       lists:map(fun({K, V}) ->
                                         {rabbit_data_coercion:to_atom(K), V}
@@ -416,12 +433,14 @@ parse_amqp10_source(Def) ->
     Address = pget(<<"src-address">>, Def),
     DeleteAfter = pget(<<"src-delete-after">>, Def, <<"never">>),
     PrefetchCount = pget(<<"src-prefetch-count">>, Def, 1000),
+    SenderCapabilities = pget(<<"src-capabilities">>, Def),
     Headers = [],
     {#{module => rabbit_amqp10_shovel,
        uris => Uris,
        source_address => Address,
        delete_after => opt_b2a(DeleteAfter),
        prefetch_count => PrefetchCount,
+       sender_capabilities => SenderCapabilities,
        consumer_args => []}, Headers}.
 
 parse_amqp091_source(Def) ->

deps/rabbitmq_shovel/test/parameters_SUITE.erl

@@ -230,6 +230,7 @@ parse_amqp10(_Config) ->
          {<<"src-address">>, <<"a-src-queue">>},
          {<<"src-delete-after">>, <<"never">>},
          {<<"src-prefetch-count">>, 30},
+         {<<"src-capabilities">>, <<"capability0">>},
 
          {<<"dest-protocol">>, <<"amqp10">>},
          {<<"dest-uri">>, <<"amqp://remotehost:5672">>},
@@ -240,7 +241,8 @@ parse_amqp10(_Config) ->
                                                <<"app-prop-value">>}]},
          {<<"dest-message-annotations">>, [{<<"some-message-ann">>,
                                             <<"message-ann-value">>}]},
-         {<<"dest-properties">>, [{<<"user_id">>, <<"some-user">>}]}
+         {<<"dest-properties">>, [{<<"user_id">>, <<"some-user">>}]},
+         {<<"dest-capabilities">>, [<<"capability1">>, <<"capability2">>]}
         ],
     ObfuscatedParams = rabbit_shovel_parameters:obfuscate_uris_in_definition(Params),
     ?assertMatch(