Add more tests

deps/amqp10_client/src/amqp10_client_session.erl

@@ -904,7 +904,7 @@ handle_link_flow(#'v1_0.flow'{delivery_count = TheirDC,
                               available = Available,
                               drain = Drain},
                  Link0 = #link{role = receiver}) ->
-    Link = case Drain andalso TheirCredit =< 0 of
+    Link = case Drain andalso TheirCredit =:= 0 of
                true ->
                    notify_credit_exhausted(Link0),
                    Link0#link{delivery_count = unpack(TheirDC),

deps/rabbit/test/amqp_client_SUITE.erl

@@ -51,6 +51,9 @@ groups() ->
        roundtrip_with_drain_classic_queue,
        roundtrip_with_drain_quorum_queue,
        roundtrip_with_drain_stream,
+       drain_many_classic_queue,
+       drain_many_quorum_queue,
+       drain_many_stream,
        amqp_stream_amqpl,
        amqp_quorum_queue_amqpl,
        message_headers_conversion,
@@ -171,6 +174,7 @@ end_per_group(_, Config) ->
 init_per_testcase(T, Config)
   when T =:= message_headers_conversion orelse
        T =:= roundtrip_with_drain_quorum_queue orelse
+       T =:= drain_many_quorum_queue orelse
        T =:= timed_get_quorum_queue orelse
        T =:= available_messages_quorum_queue ->
     case rpc(Config, rabbit_feature_flags, is_enabled, [credit_api_v2]) of
@@ -646,6 +650,69 @@ roundtrip_with_drain(Config, QueueType, QName)
     ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair),
     ok = amqp10_client:close_connection(Connection).
 
+drain_many_classic_queue(Config) ->
+    QName  = atom_to_binary(?FUNCTION_NAME),
+    drain_many(Config, <<"classic">>, QName).
+
+drain_many_quorum_queue(Config) ->
+    QName  = atom_to_binary(?FUNCTION_NAME),
+    drain_many(Config, <<"quorum">>, QName).
+
+drain_many_stream(Config) ->
+    QName  = atom_to_binary(?FUNCTION_NAME),
+    drain_many(Config, <<"stream">>, QName).
+
+drain_many(Config, QueueType, QName)
+  when is_binary(QueueType) ->
+    Address = <<"/queue/", QName/binary>>,
+    {Connection, Session, LinkPair} = init(Config),
+    QProps = #{arguments => #{<<"x-queue-type">> => {utf8, QueueType}}},
+    {ok, #{type := QueueType}} = rabbitmq_amqp_client:declare_queue(LinkPair, QName, QProps),
+    {ok, Sender} = amqp10_client:attach_sender_link(Session, <<"test-sender">>, Address),
+    wait_for_credit(Sender),
+
+    Num = 5000,
+    ok = send_messages(Sender, Num, false),
+    ok = wait_for_accepts(Num),
+
+    TerminusDurability = none,
+    Filter = consume_from_first(QueueType),
+    {ok, Receiver} = amqp10_client:attach_receiver_link(
+                       Session, <<"test-receiver">>, Address, settled,
+                       TerminusDurability, Filter),
+
+    ok = amqp10_client:flow_link_credit(Receiver, Num - 1, never, true),
+    ?assertEqual(Num - 1, count_received_messages(Receiver)),
+    flush("drained 1"),
+
+    ok = amqp10_client:flow_link_credit(Receiver, Num, never, true),
+    ?assertEqual(1, count_received_messages(Receiver)),
+    flush("drained 2"),
+
+    ok = send_messages(Sender, Num, false),
+    ok = wait_for_accepts(Num),
+    receive Unexpected -> ct:fail({unexpected, Unexpected})
+    after 10 -> ok
+    end,
+    %% Let's send 2 FLOW frames in sequence.
+    ok = amqp10_client:flow_link_credit(Receiver, Num, never, true),
+    ok = amqp10_client:flow_link_credit(Receiver, Num, never, true),
+    ?assertEqual(Num, count_received_messages(Receiver)),
+    flush("drained 3"),
+
+    ok = send_messages(Sender, 1, false),
+    ok = wait_for_accepts(1),
+    %% Our receiver shouldn't have any credit left to consume the last message.
+    receive {amqp10_msg, _, _} -> ct:fail(unexpected_delivery)
+    after 30 -> ok
+    end,
+
+    ok = amqp10_client:detach_link(Sender),
+    ok = amqp10_client:detach_link(Receiver),
+    {ok, _} = rabbitmq_amqp_client:delete_queue(LinkPair, QName),
+    ok = rabbitmq_amqp_client:detach_management_link_pair_sync(LinkPair),
+    ok = amqp10_client:close_connection(Connection).
+
 amqp_stream_amqpl(Config) ->
     amqp_amqpl(<<"stream">>, Config).