Configure capabilities on the source/target field in the ATTACH frame

Author: MarcialRosales

deps/amqp10_client/BUILD.bazel

@@ -116,6 +116,7 @@ rabbitmq_integration_suite(
     size = "medium",
     additional_beam = [
         "test/activemq_ct_helpers.beam",
+        "test/ibmmq_ct_helpers.beam",
         "test/mock_server.beam",
     ],
     data = [
@@ -139,6 +140,7 @@ eunit(
     name = "eunit",
     compiled_suites = [
         ":test_activemq_ct_helpers_beam",
+        ":test_ibmmq_ct_helpers_beam",
         ":test_mock_server_beam",
     ],
     target = ":test_erlang_app",

deps/amqp10_client/app.bzl

@@ -126,6 +126,14 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         app_name = "amqp10_client",
         erlc_opts = "//:test_erlc_opts",
     )
+    erlang_bytecode(
+        name = "test_ibmmq_ct_helpers_beam",
+        testonly = True,
+        srcs = ["test/ibmmq_ct_helpers.erl"],
+        outs = ["test/ibmmq_ct_helpers.beam"],
+        app_name = "amqp10_client",
+        erlc_opts = "//:test_erlc_opts",
+    )
     erlang_bytecode(
         name = "test_mock_server_beam",
         testonly = True,

deps/amqp10_client/src/amqp10_client.erl

@@ -166,6 +166,8 @@ end_session(Pid) ->
 %% for the link before returning.
 attach_sender_link_sync(Session, Name, Target) ->
     attach_sender_link_sync(Session, Name, Target, mixed).
+-spec attach_sender_link_sync(pid(), binary(), binary()) ->
+    {ok, link_ref()} | link_timeout.
 
 %% @doc Synchronously attach a link on 'Session'.
 %% This is a convenience function that awaits attached event
@@ -273,7 +275,8 @@ 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)
   when is_pid(Session) andalso
        is_binary(Name) andalso
        is_binary(Source) andalso

deps/amqp10_client/src/amqp10_client_session.erl

@@ -73,11 +73,14 @@
 -type rcv_settle_mode() :: first | second.
 
 -type terminus_durability() :: none | configuration | unsettled_state.
+-type terminus_capabilities() :: none | binary() | list().
 
 -type target_def() :: #{address => link_address(),
-                        durable => terminus_durability()}.
+                        durable => terminus_durability(),
+                        capabilities => terminus_capabilities()}.
 -type source_def() :: #{address => link_address(),
-                        durable => terminus_durability()}.
+                        durable => terminus_durability(),
+                        capabilities => terminus_capabilities()}.
 
 -type attach_role() :: {sender, target_def()} | {receiver, source_def(), pid()}.
 
@@ -109,6 +112,7 @@
 -export_type([snd_settle_mode/0,
               rcv_settle_mode/0,
               terminus_durability/0,
+              terminus_capabilities/0,
               attach_args/0,
               attach_role/0,
               target_def/0,
@@ -713,20 +717,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 +779,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) ->
+    {symbol, Capabilities};
+translate_terminus_capabilities(CapabilitiesList) when is_list(CapabilitiesList) ->
+    {array, symbol, [filter_capability(V) || V <- CapabilitiesList]}.
+
+filter_capability(V) when is_binary(V) ->
+    {symbol, 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,
@@ -847,7 +868,7 @@ send_attach(Send, #{name := Name, role := RoleTuple} = Args, {FromPid, _},
                             target = Target,
                             max_message_size = MaxMessageSize},
     ok = Send(Attach, State),
-
+
     Ref = make_link_ref(Role, self(), OutHandle),
     Link = #link{name = Name,
                  ref = Ref,

deps/amqp10_client/test/ibmmq_ct_helpers.erl

@@ -0,0 +1,89 @@
+%% 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-2024 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
+%%
+
+-module(ibmmq_ct_helpers).
+
+-include_lib("common_test/include/ct.hrl").
+
+-export([setup_steps/0,
+         teardown_steps/0,
+         init_config/1,
+         start_ibmmq_server/1,
+         stop_ibmmq_server/1]).
+
+setup_steps() ->
+    [fun init_config/1,
+     fun start_ibmmq_server/1
+    ].
+
+teardown_steps() ->
+    [
+     fun stop_ibmmq_server/1
+    ].
+
+init_config(Config) ->
+    NodeConfig = [{tcp_port_amqp, 5672}],
+    rabbit_ct_helpers:set_config(Config, [ {rmq_nodes, [NodeConfig]},
+                                            {rmq_hostname, "localhost"},
+                                            {tcp_hostname_amqp, "localhost"},
+                                            {sasl, {plain, <<"app">>, <<"passw0rd">>}} ]).
+
+start_ibmmq_server(Config) ->
+    IBMmqCmd = filename:join([?config(data_dir, Config), "ibmmq_runner"]),
+    Cmd = [IBMmqCmd, "start"],
+    ct:log("Running command ~p", [Cmd]),
+    case rabbit_ct_helpers:exec(Cmd, []) of
+        {ok, _} -> wait_for_ibmmq_nodes(Config);
+        Error   -> ct:pal("Error: ~tp", [Error]),
+                   {skip, "Failed to start IBM MQ"}
+    end.
+
+wait_for_ibmmq_nodes(Config) ->
+    Hostname = ?config(rmq_hostname, Config),
+    Ports = rabbit_ct_broker_helpers:get_node_configs(Config, tcp_port_amqp),
+    wait_for_ibmmq_ports(Config, Hostname, Ports).
+
+wait_for_ibmmq_ports(Config, Hostname, [Port | Rest]) ->
+    ct:log("Waiting for IBM MQ on port ~b", [Port]),
+    case wait_for_ibmmq_port(Hostname, Port, 60) of
+        ok ->
+            ct:log("IBM MQ ready on port ~b", [Port]),
+            wait_for_ibmmq_ports(Config, Hostname, Rest);
+        {error, _} ->
+            Msg = lists:flatten(
+                    io_lib:format(
+                      "Failed to start IBM MQ on port ~b; see IBM MQ logs",
+                      [Port])),
+            ct:pal(?LOW_IMPORTANCE, Msg, []),
+            {skip, Msg}
+    end;
+wait_for_ibmmq_ports(Config, _, []) ->
+    Config.
+
+wait_for_ibmmq_port(_, _, 0) ->
+    {error, econnrefused};
+wait_for_ibmmq_port(Hostname, Port, Retries) ->
+    case gen_tcp:connect(Hostname, Port, []) of
+        {ok, Connection} ->
+            gen_tcp:close(Connection),
+            ok;
+        {error, econnrefused} ->
+            timer:sleep(1000),
+            wait_for_ibmmq_port(Hostname, Port, Retries - 1);
+        Error ->
+            Error
+    end.
+
+stop_ibmmq_server(Config) ->
+    IBMmqCmd = filename:join([?config(data_dir, Config), "ibmmq_runner"]),
+    Cmd = [IBMmqCmd, "stop"],
+    ct:log("Running command ~p", [Cmd]),
+    case rabbit_ct_helpers:exec(Cmd, []) of
+        {ok, _} -> Config;
+        Error   -> ct:pal("Error: ~tp", [Error]),
+                   {skip, "Failed to stop IBM MQ"}
+    end.