Merge pull request #14360 from rabbitmq/fix-amqp10

Shovel: AMQP1.0 use prefetch-count as credit on delete-after

Author: michaelklishin

deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl

@@ -135,12 +135,7 @@ connect(Name, SndSettleMode, Uri, Postfix, Addr, Map, AttachFun) ->
 -spec init_source(state()) -> state().
 init_source(State = #{source := #{current := #{link := Link},
                                   prefetch_count := Prefetch} = Src}) ->
-    {Credit, RenewWhenBelow} = case Src of
-                                   #{delete_after := R} when is_integer(R) ->
-                                       {R, never};
-                                   #{prefetch_count := Pre} ->
-                                       {Pre, max(1, round(Prefetch/10))}
-                               end,
+    {Credit, RenewWhenBelow} = {Prefetch, max(1, round(Prefetch/10))},
     ok = amqp10_client:flow_link_credit(Link, Credit, RenewWhenBelow),
     Remaining = case Src of
                     #{delete_after := never} -> unlimited;
@@ -319,6 +314,9 @@ status(_) ->
 -spec forward(Tag :: tag(), Props :: #{atom() => any()},
               Payload :: binary(), state()) ->
     state() | {stop, any()}.
+forward(_Tag, _Props, _Payload,
+        #{source := #{remaining := 0}} = State) ->
+    State;
 forward(_Tag, _Props, _Payload,
         #{source := #{remaining_unacked := 0}} = State) ->
     State;