Test capabilities with receiver link

Author: MarcialRosales

deps/amqp10_client/test/system_SUITE.erl

@@ -52,7 +52,8 @@ groups() ->
      {mock, [], [
                  insufficient_credit,
                  incoming_heartbeat,
-                 multi_transfer_without_delivery_id
+                 multi_transfer_without_delivery_id,
+                 set_sender_capabilities
                 ]}
     ].
 
@@ -68,7 +69,7 @@ shared() ->
      subscribe_with_auto_flow_unsettled,
      outgoing_heartbeat,
      roundtrip_large_messages,
-     transfer_id_vs_delivery_id
+     transfer_id_vs_delivery_id     
     ].
 
 %% -------------------------------------------------------------------
@@ -687,6 +688,7 @@ subscribe_with_auto_flow_unsettled(Config) ->
     ok = amqp10_client:end_session(Session),
     ok = amqp10_client:close_connection(Connection).
 
+    
 insufficient_credit(Config) ->
     Hostname = ?config(mock_host, Config),
     Port = ?config(mock_port, Config),
@@ -790,6 +792,76 @@ multi_transfer_without_delivery_id(Config) ->
     ok = amqp10_client:close_connection(Connection),
     ok.
 
+set_sender_capabilities(Config) ->
+    Hostname = ?config(mock_host, Config),
+    Port = ?config(mock_port, Config),
+    
+    OpenStep = fun({0 = Ch, #'v1_0.open'{}, _Pay}) ->
+                       {Ch, [#'v1_0.open'{container_id = {utf8, <<"mock">>}}]}
+               end,
+    BeginStep = fun({1 = Ch, #'v1_0.begin'{}, _Pay}) ->
+                         {Ch, [#'v1_0.begin'{remote_channel = {ushort, 1},
+                                             next_outgoing_id = {uint, 1},
+                                             incoming_window = {uint, 1000},
+                                             outgoing_window = {uint, 1000}}
+                                             ]}
+                end,
+    AttachStep = fun({1 = Ch, #'v1_0.attach'{role = true,
+                                             name = Name,
+                                             source = #'v1_0.source'{
+                                                capabilities = {utf8, <<"capability-1">>}}}, <<>>}) ->
+                         {Ch, [#'v1_0.attach'{name = Name,
+                                              handle = {uint, 99},
+                                              initial_delivery_count = {uint, 1},
+                                              role = false}
+                              ]}
+                 end,
+
+    LinkCreditStep = fun({1 = Ch, #'v1_0.flow'{}, <<>>}) ->
+                             {Ch, {multi, [[#'v1_0.transfer'{handle = {uint, 99},
+                                                             delivery_id = {uint, 12},
+                                                             more = true},
+                                            #'v1_0.data'{content = <<"hello ">>}],
+                                           [#'v1_0.transfer'{handle = {uint, 99},
+                                                             % delivery_id can be omitted
+                                                             % for continuation frames
+                                                             delivery_id = undefined,
+                                                             settled = undefined,
+                                                             more = false},
+                                            #'v1_0.data'{content = <<"world">>}]
+                                          ]}}
+                     end,
+    Steps = [fun mock_server:recv_amqp_header_step/1,
+             fun mock_server:send_amqp_header_step/1,
+             mock_server:amqp_step(OpenStep),
+             mock_server:amqp_step(BeginStep),
+             mock_server:amqp_step(AttachStep),
+             mock_server:amqp_step(LinkCreditStep)
+            ],
+
+    ok = mock_server:set_steps(?config(mock_server, Config), Steps),
+
+    Cfg = #{address => Hostname, port => Port, sasl => none, notify => self()},
+    {ok, Connection} = amqp10_client:open_connection(Cfg),
+    {ok, Session} = amqp10_client:begin_session_sync(Connection),
+    {ok, Receiver} = amqp10_client:attach_receiver_link(Session, <<"mock1-received">>,
+                                                    <<"test">>, 
+                                                    setlled, none, 
+                                                    #{}, % filter
+                                                    #{}, % prorperties
+                                                    <<"capability-1">>),
+    amqp10_client:flow_link_credit(Receiver, 100, 50),
+    receive
+        {amqp10_msg, Receiver, _InMsg} ->
+            ok
+    after 2000 ->
+              exit(delivery_timeout)
+    end,
+
+    ok = amqp10_client:end_session(Session),
+    ok = amqp10_client:close_connection(Connection),
+    ok.
+
 outgoing_heartbeat(Config) ->
     Hostname = ?config(rmq_hostname, Config),
     Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),