Merge pull request #14708 from rabbitmq/mergify/bp/v4.2.x/pr-14704

Shovels: increase forwarded counter for AMQP1.0 (backport #14704)

Author: michaelklishin

deps/rabbitmq_shovel/src/rabbit_amqp10_shovel.erl

@@ -333,23 +333,24 @@ forward(Tag, Mc,
     State#{dest => Dst#{pending => {Pend}}};
 forward(Tag, Msg0,
         #{dest := #{current := #{link := Link},
-                    unacked := Unacked} = Dst,
+                    unacked := Unacked},
           ack_mode := AckMode} = State) ->
     OutTag = rabbit_data_coercion:to_binary(Tag),
     Msg1 = add_timestamp_header(State, add_forward_headers(State, Msg0)),
     Msg2 = mc:protocol_state(mc:convert(mc_amqp, Msg1)),
     Msg3 = amqp10_raw_msg:new(AckMode =/= on_confirm, Tag, iolist_to_binary(Msg2)),
     case send_msg(Link, Msg3) of
         ok ->
+            #{dest := Dst1} = State1 = rabbit_shovel_behaviour:incr_forwarded(State),
             rabbit_shovel_behaviour:decr_remaining_unacked(
               case AckMode of
                   no_ack ->
-                      rabbit_shovel_behaviour:decr_remaining(1, State);
+                      rabbit_shovel_behaviour:decr_remaining(1, State1);
                   on_confirm ->
-                      State#{dest => Dst#{unacked => Unacked#{OutTag => Tag}}};
+                      State1#{dest => Dst1#{unacked => Unacked#{OutTag => Tag}}};
                   on_publish ->
-                      State1 = rabbit_shovel_behaviour:ack(Tag, false, State),
-                      rabbit_shovel_behaviour:decr_remaining(1, State1)
+                      State2 = rabbit_shovel_behaviour:ack(Tag, false, State1),
+                      rabbit_shovel_behaviour:decr_remaining(1, State2)
               end);
         Stop ->
             Stop

deps/rabbitmq_shovel/test/amqp091_dynamic_SUITE.erl

@@ -11,6 +11,9 @@
 -include_lib("amqp_client/include/amqp_client.hrl").
 
 -import(rabbit_ct_helpers, [eventually/1]).
+-import(shovel_test_utils, [await_autodelete/2,
+                            invalid_param/2, invalid_param/3,
+                            valid_param/2, valid_param/3]).
 
 -compile(export_all).
 
@@ -27,7 +30,6 @@ all() ->
 groups() ->
     [
       {core_tests, [], [
-          simple,
           set_properties_using_proplist,
           set_properties_using_map,
           set_empty_properties_using_proplist,
@@ -115,21 +117,6 @@ end_per_testcase(Testcase, Config) ->
 %% -------------------------------------------------------------------
 %% Testcases.
 %% -------------------------------------------------------------------
-
-simple(Config) ->
-    Name = <<"test">>,
-    with_ch(Config,
-      fun (Ch) ->
-              shovel_test_utils:set_param(
-                Config,
-                Name, [{<<"src-queue">>,  <<"src">>},
-                             {<<"dest-queue">>, <<"dest">>}]),
-              publish_expect(Ch, <<>>, <<"src">>, <<"dest">>, <<"hello">>),
-              Status = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_shovel_status, lookup, [{<<"/">>, Name}]),
-              ?assertMatch([_|_], Status),
-              ?assertMatch(#{metrics := #{forwarded := 1}}, maps:from_list(Status))
-      end).
-
 quorum_queues(Config) ->
     with_ch(Config,
       fun (Ch) ->
@@ -814,23 +801,6 @@ expect_count(Ch, Q, M, Count) ->
      end || _ <- lists:seq(1, Count)],
     expect_empty(Ch, Q).
 
-invalid_param(Config, Value, User) ->
-    {error_string, _} = rabbit_ct_broker_helpers:rpc(Config, 0,
-      rabbit_runtime_parameters, set,
-      [<<"/">>, <<"shovel">>, <<"invalid">>, Value, User]).
-
-valid_param(Config, Value, User) ->
-    rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, valid_param1, [Config, Value, User]).
-
-valid_param1(_Config, Value, User) ->
-    ok = rabbit_runtime_parameters:set(
-           <<"/">>, <<"shovel">>, <<"name">>, Value, User),
-    ok = rabbit_runtime_parameters:clear(<<"/">>, <<"shovel">>, <<"name">>, <<"acting-user">>).
-
-invalid_param(Config, Value) -> invalid_param(Config, Value, none).
-valid_param(Config, Value) -> valid_param(Config, Value, none).
-
 lookup_user(Config, Name) ->
     {ok, User} = rabbit_ct_broker_helpers:rpc(Config, 0,
       rabbit_access_control, check_user_login, [Name, []]),
@@ -849,23 +819,6 @@ cleanup1(_Config) ->
     [rabbit_amqqueue:delete(Q, false, false, <<"acting-user">>)
      || Q <- rabbit_amqqueue:list()].
 
-await_autodelete(Config, Name) ->
-    rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, await_autodelete1, [Config, Name]).
-
-await_autodelete1(_Config, Name) ->
-    shovel_test_utils:await(
-      fun () -> not lists:member(Name, shovels_from_parameters()) end),
-    shovel_test_utils:await(
-      fun () ->
-              not lists:member(Name,
-                               shovel_test_utils:shovels_from_status())
-      end).
-
-shovels_from_parameters() ->
-    L = rabbit_runtime_parameters:list(<<"/">>, <<"shovel">>),
-    [rabbit_misc:pget(name, Shovel) || Shovel <- L].
-
 set_default_credit(Config, Value) ->
     Key = credit_flow_default_credit,
     OrigValue = rabbit_ct_broker_helpers:rpc(Config, persistent_term, get, [Key]),