|
| 1 | +%% This Source Code Form is subject to the terms of the Mozilla Public |
| 2 | +%% License, v. 2.0. If a copy of the MPL was not distributed with this |
| 3 | +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. |
| 4 | +%% |
| 5 | +%% Copyright (c) 2016-2023 VMware, Inc. or its affiliates. All rights reserved. |
| 6 | + |
| 7 | +-module(ff_SUITE). |
| 8 | + |
| 9 | +-compile([export_all, nowarn_export_all]). |
| 10 | + |
| 11 | +-include_lib("common_test/include/ct.hrl"). |
| 12 | +-include_lib("eunit/include/eunit.hrl"). |
| 13 | +-include_lib("amqp_client/include/amqp_client.hrl"). |
| 14 | + |
| 15 | +all() -> |
| 16 | + [ |
| 17 | + {group, cluster_size_1} |
| 18 | + ]. |
| 19 | + |
| 20 | +groups() -> |
| 21 | + [ |
| 22 | + {cluster_size_1, [], |
| 23 | + [credit_api_v2]} |
| 24 | + ]. |
| 25 | + |
| 26 | +suite() -> |
| 27 | + [ |
| 28 | + {timetrap, {minutes, 10}} |
| 29 | + ]. |
| 30 | + |
| 31 | +init_per_suite(Config) -> |
| 32 | + {ok, _} = application:ensure_all_started(amqp10_client), |
| 33 | + rabbit_ct_helpers:log_environment(), |
| 34 | + rabbit_ct_helpers:run_setup_steps(Config, []). |
| 35 | + |
| 36 | +end_per_suite(Config) -> |
| 37 | + rabbit_ct_helpers:run_teardown_steps(Config). |
| 38 | + |
| 39 | +init_per_group(_Group, Config0) -> |
| 40 | + Config = rabbit_ct_helpers:merge_app_env( |
| 41 | + Config0, {rabbit, [{forced_feature_flags_on_init, []}]}), |
| 42 | + rabbit_ct_helpers:run_steps(Config, |
| 43 | + rabbit_ct_broker_helpers:setup_steps() ++ |
| 44 | + rabbit_ct_client_helpers:setup_steps()). |
| 45 | + |
| 46 | +end_per_group(_Group, Config) -> |
| 47 | + rabbit_ct_helpers:run_steps(Config, |
| 48 | + rabbit_ct_client_helpers:teardown_steps() ++ |
| 49 | + rabbit_ct_broker_helpers:teardown_steps()). |
| 50 | + |
| 51 | +init_per_testcase(TestCase, Config) -> |
| 52 | + case rabbit_ct_broker_helpers:is_feature_flag_supported(Config, TestCase) of |
| 53 | + true -> |
| 54 | + ?assertNot(rabbit_ct_broker_helpers:is_feature_flag_enabled(Config, TestCase)), |
| 55 | + Config; |
| 56 | + false -> |
| 57 | + {skip, io_lib:format("feature flag ~s is unsupported", [TestCase])} |
| 58 | + end. |
| 59 | + |
| 60 | +end_per_testcase(_TestCase, Config) -> |
| 61 | + Config. |
| 62 | + |
| 63 | +credit_api_v2(Config) -> |
| 64 | + CQ = <<"classic queue">>, |
| 65 | + QQ = <<"quorum queue">>, |
| 66 | + CQAddr = <<"/amq/queue/", CQ/binary>>, |
| 67 | + QQAddr = <<"/amq/queue/", QQ/binary>>, |
| 68 | + |
| 69 | + Ch = rabbit_ct_client_helpers:open_channel(Config), |
| 70 | + #'queue.declare_ok'{} = amqp_channel:call(Ch, #'queue.declare'{queue = CQ}), |
| 71 | + #'queue.declare_ok'{} = amqp_channel:call( |
| 72 | + Ch, #'queue.declare'{ |
| 73 | + queue = QQ, |
| 74 | + durable = true, |
| 75 | + arguments = [{<<"x-queue-type">>, longstr, <<"quorum">>}]}), |
| 76 | + ok = rabbit_ct_client_helpers:close_channel(Ch), |
| 77 | + |
| 78 | + Host = ?config(rmq_hostname, Config), |
| 79 | + Port = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp), |
| 80 | + OpnConf = #{address => Host, |
| 81 | + port => Port, |
| 82 | + container_id => <<"my container">>, |
| 83 | + sasl => {plain, <<"guest">>, <<"guest">>}}, |
| 84 | + {ok, Connection} = amqp10_client:open_connection(OpnConf), |
| 85 | + {ok, Session} = amqp10_client:begin_session_sync(Connection), |
| 86 | + |
| 87 | + {ok, CQSender} = amqp10_client:attach_sender_link(Session, <<"cq sender">>, CQAddr), |
| 88 | + {ok, QQSender} = amqp10_client:attach_sender_link(Session, <<"qq sender">>, QQAddr), |
| 89 | + receive {amqp10_event, {link, CQSender, credited}} -> ok |
| 90 | + after 5000 -> ct:fail(credited_timeout) |
| 91 | + end, |
| 92 | + receive {amqp10_event, {link, QQSender, credited}} -> ok |
| 93 | + after 5000 -> ct:fail(credited_timeout) |
| 94 | + end, |
| 95 | + |
| 96 | + %% Send 40 messages to each queue. |
| 97 | + NumMsgs = 40, |
| 98 | + [begin |
| 99 | + Bin = integer_to_binary(N), |
| 100 | + ok = amqp10_client:send_msg(CQSender, amqp10_msg:new(Bin, Bin, true)), |
| 101 | + ok = amqp10_client:send_msg(QQSender, amqp10_msg:new(Bin, Bin, true)) |
| 102 | + end || N <- lists:seq(1, NumMsgs)], |
| 103 | + ok = amqp10_client:detach_link(CQSender), |
| 104 | + ok = amqp10_client:detach_link(QQSender), |
| 105 | + |
| 106 | + %% Consume with credit API v1 |
| 107 | + CQAttachArgs = #{handle => 300, |
| 108 | + name => <<"cq receiver 1">>, |
| 109 | + role => {receiver, #{address => CQAddr, |
| 110 | + durable => configuration}, self()}, |
| 111 | + snd_settle_mode => unsettled, |
| 112 | + rcv_settle_mode => first, |
| 113 | + filter => #{}}, |
| 114 | + {ok, CQReceiver1} = amqp10_client:attach_link(Session, CQAttachArgs), |
| 115 | + QQAttachArgs = #{handle => 400, |
| 116 | + name => <<"qq receiver 1">>, |
| 117 | + role => {receiver, #{address => QQAddr, |
| 118 | + durable => configuration}, self()}, |
| 119 | + snd_settle_mode => unsettled, |
| 120 | + rcv_settle_mode => first, |
| 121 | + filter => #{}}, |
| 122 | + {ok, QQReceiver1} = amqp10_client:attach_link(Session, QQAttachArgs), |
| 123 | + |
| 124 | + ok = consume_and_accept(10, CQReceiver1, Session), |
| 125 | + ok = consume_and_accept(10, QQReceiver1, Session), |
| 126 | + |
| 127 | + ?assertEqual(ok, |
| 128 | + rabbit_ct_broker_helpers:enable_feature_flag(Config, ?FUNCTION_NAME)), |
| 129 | + flush(enabled_feature_flag), |
| 130 | + |
| 131 | + %% Consume with credit API v2 |
| 132 | + {ok, CQReceiver2} = amqp10_client:attach_receiver_link( |
| 133 | + Session, <<"cq receiver 2">>, CQAddr, unsettled), |
| 134 | + {ok, QQReceiver2} = amqp10_client:attach_receiver_link( |
| 135 | + Session, <<"qq receiver 2">>, QQAddr, unsettled), |
| 136 | + ok = consume_and_accept(10, CQReceiver2, Session), |
| 137 | + ok = consume_and_accept(10, QQReceiver2, Session), |
| 138 | + |
| 139 | + %% Consume via with credit API v1 |
| 140 | + ok = consume_and_accept(10, CQReceiver1, Session), |
| 141 | + ok = consume_and_accept(10, QQReceiver1, Session), |
| 142 | + |
| 143 | + %% Detach the credit API v1 links and attach with the same output handle. |
| 144 | + ok = detach_sync(CQReceiver1), |
| 145 | + ok = detach_sync(QQReceiver1), |
| 146 | + {ok, CQReceiver3} = amqp10_client:attach_link(Session, CQAttachArgs), |
| 147 | + {ok, QQReceiver3} = amqp10_client:attach_link(Session, QQAttachArgs), |
| 148 | + |
| 149 | + %% The new links should use credit API v2 |
| 150 | + ok = consume_and_accept(10, CQReceiver3, Session), |
| 151 | + ok = consume_and_accept(10, QQReceiver3, Session), |
| 152 | + |
| 153 | + flush(pre_drain), |
| 154 | + %% Draining should also work. |
| 155 | + ok = amqp10_client:flow_link_credit(CQReceiver3, 10, never, true), |
| 156 | + receive {amqp10_event, {link, CQReceiver3, credit_exhausted}} -> ok |
| 157 | + after 5000 -> ct:fail({missing_credit_exhausted, ?LINE}) |
| 158 | + end, |
| 159 | + receive Unexpected1 -> ct:fail({unexpected, ?LINE, Unexpected1}) |
| 160 | + after 20 -> ok |
| 161 | + end, |
| 162 | + |
| 163 | + ok = amqp10_client:flow_link_credit(QQReceiver3, 10, never, true), |
| 164 | + receive {amqp10_event, {link, QQReceiver3, credit_exhausted}} -> ok |
| 165 | + after 5000 -> ct:fail({missing_credit_exhausted, ?LINE}) |
| 166 | + end, |
| 167 | + receive Unexpected2 -> ct:fail({unexpected, ?LINE, Unexpected2}) |
| 168 | + after 20 -> ok |
| 169 | + end, |
| 170 | + |
| 171 | + ok = detach_sync(CQReceiver2), |
| 172 | + ok = detach_sync(QQReceiver2), |
| 173 | + ok = detach_sync(CQReceiver3), |
| 174 | + ok = detach_sync(QQReceiver3), |
| 175 | + ok = amqp10_client:end_session(Session), |
| 176 | + receive {amqp10_event, {session, Session, {ended, _}}} -> ok |
| 177 | + after 5000 -> ct:fail(missing_ended) |
| 178 | + end, |
| 179 | + ok = amqp10_client:close_connection(Connection), |
| 180 | + receive {amqp10_event, {connection, Connection, {closed, normal}}} -> ok |
| 181 | + after 5000 -> ct:fail(missing_closed) |
| 182 | + end. |
| 183 | + |
| 184 | +consume_and_accept(NumMsgs, Receiver, Session) -> |
| 185 | + ok = amqp10_client:flow_link_credit(Receiver, NumMsgs, never), |
| 186 | + Msgs = receive_messages(Receiver, NumMsgs), |
| 187 | + ok = amqp10_client_session:disposition( |
| 188 | + Session, |
| 189 | + receiver, |
| 190 | + amqp10_msg:delivery_id(hd(Msgs)), |
| 191 | + amqp10_msg:delivery_id(lists:last(Msgs)), |
| 192 | + true, |
| 193 | + accepted). |
| 194 | + |
| 195 | +receive_messages(Receiver, N) -> |
| 196 | + receive_messages0(Receiver, N, []). |
| 197 | + |
| 198 | +receive_messages0(_Receiver, 0, Acc) -> |
| 199 | + lists:reverse(Acc); |
| 200 | +receive_messages0(Receiver, N, Acc) -> |
| 201 | + receive |
| 202 | + {amqp10_msg, Receiver, Msg} -> |
| 203 | + receive_messages0(Receiver, N - 1, [Msg | Acc]) |
| 204 | + after 5000 -> |
| 205 | + exit({timeout, {num_received, length(Acc)}, {num_missing, N}}) |
| 206 | + end. |
| 207 | + |
| 208 | +detach_sync(Receiver) -> |
| 209 | + ok = amqp10_client:detach_link(Receiver), |
| 210 | + receive {amqp10_event, {link, Receiver, {detached, normal}}} -> ok |
| 211 | + after 5000 -> ct:fail({missing_detached, Receiver}) |
| 212 | + end. |
| 213 | + |
| 214 | +flush(Prefix) -> |
| 215 | + receive |
| 216 | + Msg -> |
| 217 | + ct:pal("~ts flushed: ~p~n", [Prefix, Msg]), |
| 218 | + flush(Prefix) |
| 219 | + after 1 -> |
| 220 | + ok |
| 221 | + end. |
0 commit comments