Cleanup compat code

Author: ansd

deps/rabbit/src/mc_amqpl.erl

@@ -58,12 +58,12 @@ init(#content{} = Content0) ->
     Content = strip_header(Content1, ?DELETED_HEADER),
     {Content, Anns}.
 
-convert_from(mc_amqp, Sections, Env) ->
-    {H, MAnn, Prop, AProp, BodyRev, Footer} =
+convert_from(mc_amqp, Sections, _Env) ->
+    {H, MAnn, Prop, AProp, BodyRev} =
     lists:foldl(
       fun(#'v1_0.header'{} = S, Acc) ->
               setelement(1, Acc, S);
-         (_Ignore = #'v1_0.delivery_annotations'{}, Acc) ->
+         (#'v1_0.delivery_annotations'{}, Acc) ->
               Acc;
          (#'v1_0.message_annotations'{} = S, Acc) ->
               setelement(2, Acc, S);
@@ -81,10 +81,10 @@ convert_from(mc_amqp, Sections, Env) ->
               %% assertions
               [] = element(5, Acc),
               setelement(5, Acc, Body);
-         (#'v1_0.footer'{} = S, Acc) ->
-              setelement(6, Acc, S)
+         (#'v1_0.footer'{}, Acc) ->
+              Acc
       end,
-      {undefined, undefined, undefined, undefined, [], undefined},
+      {undefined, undefined, undefined, undefined, []},
       Sections),
 
     {PFR, Type0} = case BodyRev of
@@ -181,42 +181,7 @@ convert_from(mc_amqp, Sections, Env) ->
                                        false
                                end, MA),
     {Headers1, MsgId091} = message_id(MsgId, <<"x-message-id">>, Headers0),
-    {Headers2, CorrId091} = message_id(CorrId, <<"x-correlation-id">>, Headers1),
-
-    Headers = case Env of
-                  #{'rabbitmq_4.0.0' := false} ->
-                      Headers3 = case AProp of
-                                     undefined ->
-                                         Headers2;
-                                     #'v1_0.application_properties'{} ->
-                                         APropBin = amqp_encoded_binary(AProp),
-                                         [{?AMQP10_APP_PROPERTIES_HEADER, longstr, APropBin} | Headers2]
-                                 end,
-                      Headers4 = case Prop of
-                                     undefined ->
-                                         Headers3;
-                                     #'v1_0.properties'{} ->
-                                         PropBin = amqp_encoded_binary(Prop),
-                                         [{?AMQP10_PROPERTIES_HEADER, longstr, PropBin} | Headers3]
-                                 end,
-                      Headers5 = case MAnn of
-                                     undefined ->
-                                         Headers4;
-                                     #'v1_0.message_annotations'{} ->
-                                         MAnnBin = amqp_encoded_binary(MAnn),
-                                         [{?AMQP10_MESSAGE_ANNOTATIONS_HEADER, longstr, MAnnBin} | Headers4]
-                                 end,
-                      Headers6 = case Footer of
-                                     undefined ->
-                                         Headers5;
-                                     #'v1_0.footer'{} ->
-                                         FootBin = amqp_encoded_binary(Footer),
-                                         [{?AMQP10_FOOTER, longstr, FootBin} | Headers5]
-                                 end,
-                      Headers6;
-                  _ ->
-                      Headers2
-              end,
+    {Headers, CorrId091} = message_id(CorrId, <<"x-correlation-id">>, Headers1),
 
     UserId1 = unwrap(UserId0),
     %% user-id is a binary type so we need to validate

deps/rabbit/src/rabbit_amqp_session.erl

@@ -1357,7 +1357,7 @@ handle_call({basic_get, ChPid, NoAck, LimiterPid}, _From,
     end;
 
 handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
-             ModeOrPrefetch, ConsumerTag, ExclusiveConsume, Args, OkMsg, ActingUser},
+             Mode, ConsumerTag, ExclusiveConsume, Args, OkMsg, ActingUser},
             _From, State = #q{consumers = Consumers,
                               active_consumer = Holder,
                               single_active_consumer_on = SingleActiveConsumerOn}) ->
@@ -1369,7 +1369,7 @@ handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
                 false ->
                     Consumers1 = rabbit_queue_consumers:add(
                                    ChPid, ConsumerTag, NoAck,
-                                   LimiterPid, LimiterActive, ModeOrPrefetch,
+                                   LimiterPid, LimiterActive, Mode,
                                    Args, ActingUser, Consumers),
                     case Holder of
                         none ->
@@ -1388,7 +1388,7 @@ handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
               ok     ->
                     Consumers1 = rabbit_queue_consumers:add(
                                    ChPid, ConsumerTag, NoAck,
-                                   LimiterPid, LimiterActive, ModeOrPrefetch,
+                                   LimiterPid, LimiterActive, Mode,
                                    Args, ActingUser, Consumers),
                     ExclusiveConsumer =
                         if ExclusiveConsume -> {ChPid, ConsumerTag};
@@ -1416,7 +1416,7 @@ handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
                     {false, _} ->
                        {true, up}
                 end,
-                PrefetchCount = rabbit_queue_consumers:parse_prefetch_count(ModeOrPrefetch),
+                PrefetchCount = rabbit_queue_consumers:parse_prefetch_count(Mode),
                 rabbit_core_metrics:consumer_created(
                 ChPid, ConsumerTag, ExclusiveConsume, AckRequired, QName,
                 PrefetchCount, ConsumerIsActive, ActivityStatus, Args),
@@ -1427,13 +1427,6 @@ handle_call({basic_consume, NoAck, ChPid, LimiterPid, LimiterActive,
             reply(ok, run_message_queue(false, State1))
     end;
 
-handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg, ActingUser}, From, State) ->
-    handle_call({stop_consumer, #{pid => ChPid,
-                                  consumer_tag => ConsumerTag,
-                                  ok_msg => OkMsg,
-                                  user => ActingUser}},
-                From, State);
-
 handle_call({stop_consumer, #{pid := ChPid,
                               consumer_tag := ConsumerTag,
                               user := ActingUser} = Spec},
@@ -1597,16 +1590,6 @@ handle_cast({deactivate_limit, ChPid}, State) ->
     noreply(possibly_unblock(rabbit_queue_consumers:deactivate_limit_fun(),
                              ChPid, State));
 
-handle_cast({credit, SessionPid, CTag, Credit, Drain},
-            #q{q = Q,
-               backing_queue = BQ,
-               backing_queue_state = BQS0} = State) ->
-    %% Credit API v1.
-    %% Delete this function clause when feature flag rabbitmq_4.0.0 becomes required.
-    %% Behave like non-native AMQP 1.0: Send send_credit_reply before deliveries.
-    rabbit_classic_queue:send_credit_reply_credit_api_v1(
-      SessionPid, amqqueue:get_name(Q), BQ:len(BQS0)),
-    handle_cast({credit, SessionPid, CTag, credit_api_v1, Credit, Drain}, State);
 handle_cast({credit, SessionPid, CTag, DeliveryCountRcv, Credit, Drain},
             #q{consumers = Consumers0,
                q = Q} = State0) ->
@@ -1622,25 +1605,15 @@ handle_cast({credit, SessionPid, CTag, DeliveryCountRcv, Credit, Drain},
                 run_message_queue(true, State1)
         end,
     case rabbit_queue_consumers:get_link_state(SessionPid, CTag) of
-        {credit_api_v1, PostCred}
-          when Drain andalso
-               is_integer(PostCred) andalso PostCred > 0 ->
-            %% credit API v1
-            rabbit_queue_consumers:drained(credit_api_v1, SessionPid, CTag),
-            rabbit_classic_queue:send_drained_credit_api_v1(SessionPid, QName, CTag, PostCred);
         {PostDeliveryCountSnd, PostCred}
-          when is_integer(PostDeliveryCountSnd) andalso
-               Drain andalso
+          when Drain andalso
                is_integer(PostCred) andalso PostCred > 0 ->
-            %% credit API v2
             AdvancedDeliveryCount = serial_number:add(PostDeliveryCountSnd, PostCred),
             rabbit_queue_consumers:drained(AdvancedDeliveryCount, SessionPid, CTag),
             Avail = BQ:len(PostBQS),
             rabbit_classic_queue:send_credit_reply(
               SessionPid, QName, CTag, AdvancedDeliveryCount, 0, Avail, Drain);
-        {PostDeliveryCountSnd, PostCred}
-          when is_integer(PostDeliveryCountSnd) ->
-            %% credit API v2
+        {PostDeliveryCountSnd, PostCred} ->
             Avail = BQ:len(PostBQS),
             rabbit_classic_queue:send_credit_reply(
               SessionPid, QName, CTag, PostDeliveryCountSnd, PostCred, Avail, Drain);

deps/rabbit/src/rabbit_amqqueue_process.erl

@@ -55,7 +55,6 @@
          supports_stateful_delivery/0,
          deliver/3,
          settle/5,
-         credit_v1/5,
          credit/6,
          dequeue/5,
          info/2,
@@ -72,8 +71,6 @@
 -export([confirm_to_sender/3,
          send_rejection/3,
          deliver_to_consumer/5,
-         send_credit_reply_credit_api_v1/3,
-         send_drained_credit_api_v1/4,
          send_credit_reply/7]).
 
 -export([policy_apply_to_name/0,
@@ -303,14 +300,13 @@ consume(Q, Spec, State0) when ?amqqueue_is_classic(Q) ->
       mode := Mode,
       consumer_tag := ConsumerTag,
       exclusive_consume := ExclusiveConsume,
-      args := Args0,
+      args := Args,
       ok_msg := OkMsg,
       acting_user :=  ActingUser} = Spec,
-    {ModeOrPrefetch, Args} = consume_backwards_compat(Mode, Args0),
     case delegate:invoke(QPid,
                          {gen_server2, call,
                           [{basic_consume, NoAck, ChPid, LimiterPid,
-                            LimiterActive, ModeOrPrefetch, ConsumerTag,
+                            LimiterActive, Mode, ConsumerTag,
                             ExclusiveConsume, Args, OkMsg, ActingUser},
                            infinity]}) of
         ok ->
@@ -324,34 +320,9 @@ consume(Q, Spec, State0) when ?amqqueue_is_classic(Q) ->
              [rabbit_misc:rs(QRef), Reason]}
     end.
 
-%% Delete this function when feature flag rabbitmq_4.0.0 becomes required.
-consume_backwards_compat({simple_prefetch, PrefetchCount} = Mode, Args) ->
-    case rabbit_feature_flags:is_enabled('rabbitmq_4.0.0') of
-        true -> {Mode, Args};
-        false -> {PrefetchCount, Args}
-    end;
-consume_backwards_compat({credited, InitialDeliveryCount} = Mode, Args)
-  when is_integer(InitialDeliveryCount) ->
-    %% credit API v2
-    {Mode, Args};
-consume_backwards_compat({credited, credit_api_v1}, Args) ->
-    %% credit API v1
-    {_PrefetchCount = 0,
-     [{<<"x-credit">>, table, [{<<"credit">>, long, 0},
-                               {<<"drain">>,  bool, false}]} | Args]}.
-
 cancel(Q, Spec, State) ->
-    %% Cancel API v2 reuses feature flag rabbitmq_4.0.0.
-    Request = case rabbit_feature_flags:is_enabled('rabbitmq_4.0.0') of
-                  true ->
-                      {stop_consumer, Spec#{pid => self()}};
-                  false ->
-                      #{consumer_tag := ConsumerTag,
-                        user := ActingUser} = Spec,
-                      OkMsg = maps:get(ok_msg, Spec, undefined),
-                      {basic_cancel, self(), ConsumerTag, OkMsg, ActingUser}
-              end,
     Pid = amqqueue:get_pid(Q),
+    Request = {stop_consumer, Spec#{pid => self()}},
     case delegate:invoke(Pid, {gen_server2, call, [Request, infinity]}) of
         ok ->
             {ok, State#?STATE{pid = Pid}};
@@ -381,11 +352,6 @@ settle(_QName, Op, _CTag, MsgIds, State = #?STATE{pid = Pid}) ->
     delegate:invoke_no_result(Pid, {gen_server2, cast, [Arg]}),
     {State, []}.
 
-credit_v1(_QName, Ctag, LinkCreditSnd, Drain, #?STATE{pid = QPid} = State) ->
-    Request = {credit, self(), Ctag, LinkCreditSnd, Drain},
-    delegate:invoke_no_result(QPid, {gen_server2, cast, [Request]}),
-    {State, []}.
-
 credit(_QName, Ctag, DeliveryCountRcv, LinkCreditRcv, Drain, #?STATE{pid = QPid} = State) ->
     Request = {credit, self(), Ctag, DeliveryCountRcv, LinkCreditRcv, Drain},
     delegate:invoke_no_result(QPid, {gen_server2, cast, [Request]}),
@@ -448,11 +414,6 @@ handle_event(QName, {down, Pid, Info}, #?STATE{monitored = Monitored,
     end;
 handle_event(_QName, Action, State)
   when element(1, Action) =:= credit_reply ->
-    {ok, State, [Action]};
-handle_event(_QName, {send_drained, {Ctag, Credit}}, State) ->
-    %% This function clause should be deleted when feature flag
-    %% rabbitmq_4.0.0 becomes required.
-    Action = {credit_reply_v1, Ctag, Credit, _Available = 0, _Drain = true},
     {ok, State, [Action]}.
 
 settlement_action(_Type, _QRef, [], Acc) ->
@@ -661,11 +622,8 @@ capabilities() ->
                           <<"x-dead-letter-routing-key">>, <<"x-max-length">>,
                           <<"x-max-length-bytes">>, <<"x-max-priority">>,
                           <<"x-overflow">>, <<"x-queue-mode">>, <<"x-queue-version">>,
-                          <<"x-single-active-consumer">>, <<"x-queue-type">>, <<"x-queue-master-locator">>]
-                          ++ case rabbit_feature_flags:is_enabled('rabbitmq_4.0.0') of
-                                 true -> [<<"x-queue-leader-locator">>];
-                                 false -> []
-                             end,
+                          <<"x-single-active-consumer">>, <<"x-queue-type">>,
+                          <<"x-queue-master-locator">>, <<"x-queue-leader-locator">>],
       consumer_arguments => [<<"x-priority">>],
       server_named => true,
       rebalance_module => undefined,
@@ -742,16 +700,6 @@ deliver_to_consumer(Pid, QName, CTag, AckRequired, Message) ->
     Evt = {deliver, CTag, AckRequired, [Message]},
     send_queue_event(Pid, QName, Evt).
 
-%% Delete this function when feature flag rabbitmq_4.0.0 becomes required.
-send_credit_reply_credit_api_v1(Pid, QName, Available) ->
-    Evt = {send_credit_reply, Available},
-    send_queue_event(Pid, QName, Evt).
-
-%% Delete this function when feature flag rabbitmq_4.0.0 becomes required.
-send_drained_credit_api_v1(Pid, QName, Ctag, Credit) ->
-    Evt = {send_drained, {Ctag, Credit}},
-    send_queue_event(Pid, QName, Evt).
-
 send_credit_reply(Pid, QName, Ctag, DeliveryCount, Credit, Available, Drain) ->
     Evt = {credit_reply, Ctag, DeliveryCount, Credit, Available, Drain},
     send_queue_event(Pid, QName, Evt).