Improve readability

Author: ansd

deps/rabbit/src/rabbit_core_ff.erl

@@ -195,7 +195,8 @@
 
 -rabbit_feature_flag(
    {quorum_queues_v4,
-    #{desc => "Unlocks QQ v4 goodies",
+    #{desc => "Unlocks quorum queue v4 features",
+      doc_url => "https://github.com/rabbitmq/rabbitmq-server/pull/10637",
       stability => stable,
       depends_on => [quorum_queue,
                      credit_api_v2]

deps/rabbit/src/rabbit_fifo.erl

@@ -1400,10 +1400,10 @@ cancel_consumer0(Meta, ConsumerKey,
 activate_next_consumer({State, Effects}) ->
     activate_next_consumer(State, Effects).
 
-activate_next_consumer(#?STATE{cfg = #cfg{consumer_strategy = competing}} = State0,
-                       Effects0) ->
-    {State0, Effects0};
-activate_next_consumer(#?STATE{consumers = Cons,
+activate_next_consumer(#?STATE{cfg = #cfg{consumer_strategy = competing}} = State,
+                       Effects) ->
+    {State, Effects};
+activate_next_consumer(#?STATE{consumers = Cons0,
                                waiting_consumers = Waiting0} = State0,
                        Effects0) ->
     %% invariant, the waiting list always need to be sorted by consumers that are
@@ -1416,11 +1416,11 @@ activate_next_consumer(#?STATE{consumers = Cons,
                 undefined
         end,
 
-    case {active_consumer(Cons), NextConsumer} of
+    case {active_consumer(Cons0), NextConsumer} of
         {undefined, {NextCKey, #consumer{cfg = NextCCfg} = NextC}} ->
             Remaining = tl(Waiting0),
             %% TODO: can this happen?
-            Consumer = case maps:get(NextCKey, Cons, undefined) of
+            Consumer = case maps:get(NextCKey, Cons0, undefined) of
                            undefined ->
                                NextC;
                            Existing ->
@@ -1433,7 +1433,7 @@ activate_next_consumer(#?STATE{consumers = Cons,
             ServiceQueue1 = maybe_queue_consumer(NextCKey,
                                                  Consumer,
                                                  ServiceQueue),
-            State = State0#?STATE{consumers = Cons#{NextCKey => Consumer},
+            State = State0#?STATE{consumers = Cons0#{NextCKey => Consumer},
                                   service_queue = ServiceQueue1,
                                   waiting_consumers = Remaining},
             Effects = consumer_update_active_effects(State, Consumer,
@@ -1452,11 +1452,12 @@ activate_next_consumer(#?STATE{consumers = Cons,
             ServiceQueue1 = maybe_queue_consumer(NextCKey,
                                                  Consumer,
                                                  ServiceQueue),
-            State = State0#?STATE{consumers = maps:remove(ActiveCKey,
-                                                          Cons#{NextCKey => Consumer}),
+            Cons1 = Cons0#{NextCKey => Consumer},
+            Cons = maps:remove(ActiveCKey, Cons1),
+            Waiting = add_waiting({ActiveCKey, Active}, Remaining),
+            State = State0#?STATE{consumers = Cons,
                                   service_queue = ServiceQueue1,
-                                  waiting_consumers =
-                                      add_waiting({ActiveCKey, Active}, Remaining)},
+                                  waiting_consumers = Waiting},
             Effects = consumer_update_active_effects(State, Consumer,
                                                      true, single_active,
                                                      Effects0),
@@ -1466,9 +1467,10 @@ activate_next_consumer(#?STATE{consumers = Cons,
           when WaitingPriority > ActivePriority ->
             %% A higher priority consumer has attached but the current one has
             %% pending messages
-            {State0#?STATE{consumers =
-                           Cons#{ActiveCKey => Active#consumer{status = fading}}},
-             Effects0};
+            Cons = maps:update(ActiveCKey,
+                               Active#consumer{status = fading},
+                               Cons0),
+            {State0#?STATE{consumers = Cons}, Effects0};
         _ ->
             %% no activation
             {State0, Effects0}
@@ -1504,10 +1506,10 @@ maybe_return_all(#{system_time := Ts} = Meta, ConsumerKey,
                                  status = cancelled},
                S0), Effects0};
         _ ->
-            {S1, Effects1} = return_all(Meta, S0, Effects0, ConsumerKey, Consumer),
+            {S1, Effects} = return_all(Meta, S0, Effects0, ConsumerKey, Consumer),
             {S1#?STATE{consumers = maps:remove(ConsumerKey, S1#?STATE.consumers),
                        last_active = Ts},
-             Effects1}
+             Effects}
     end.
 
 apply_enqueue(#{index := RaftIdx,
@@ -1743,8 +1745,7 @@ update_header(Key, UpdateFun, Default, ?TUPLE(Size, Expiry))
     update_header(Key, UpdateFun, Default, #{size => Size,
                                              expiry => Expiry});
 update_header(Key, UpdateFun, Default, Header)
-  when is_map(Header) andalso
-       is_map_key(size, Header) ->
+  when is_map_key(size, Header) ->
     maps:update_with(Key, UpdateFun, Default, Header).
 
 get_msg_header(?MSG(_Idx, Header)) ->
@@ -2172,24 +2173,24 @@ update_consumer(Meta, ConsumerKey, {Tag, Pid}, ConsumerMeta,
                 {Life, Mode} = Spec, Priority,
                 #?STATE{cfg = #cfg{consumer_strategy = single_active},
                         consumers = Cons0,
-                        waiting_consumers = Waiting,
-                        service_queue = _ServiceQueue0} = State0) ->
+                        waiting_consumers = Waiting0,
+                        service_queue = _ServiceQueue0} = State) ->
     %% if it is the current active consumer, just update
     %% if it is a cancelled active consumer, add to waiting unless it is the only
     %% one, then merge
     case active_consumer(Cons0) of
         {ConsumerKey, #consumer{status = up} = Consumer0} ->
             Consumer = merge_consumer(Meta, Consumer0, ConsumerMeta,
                                       Spec, Priority),
-            {Consumer, update_or_remove_con(Meta, ConsumerKey, Consumer, State0)};
+            {Consumer, update_or_remove_con(Meta, ConsumerKey, Consumer, State)};
         undefined when is_map_key(ConsumerKey, Cons0) ->
             %% there is no active consumer and the current consumer is in the
             %% consumers map and thus must be cancelled, in this case we can just
             %% merge and effectively make this the current active one
             Consumer0 = maps:get(ConsumerKey, Cons0),
             Consumer = merge_consumer(Meta, Consumer0, ConsumerMeta,
                                       Spec, Priority),
-            {Consumer, update_or_remove_con(Meta, ConsumerKey, Consumer, State0)};
+            {Consumer, update_or_remove_con(Meta, ConsumerKey, Consumer, State)};
         _ ->
             %% add as a new waiting consumer
             Credit = included_credit(Mode),
@@ -2202,9 +2203,8 @@ update_consumer(Meta, ConsumerKey, {Tag, Pid}, ConsumerMeta,
                                                      credit_mode = Mode},
                                  credit = Credit,
                                  delivery_count = DeliveryCount},
-            {Consumer,
-             State0#?STATE{waiting_consumers =
-                           add_waiting({ConsumerKey, Consumer}, Waiting)}}
+            Waiting = add_waiting({ConsumerKey, Consumer}, Waiting0),
+            {Consumer, State#?STATE{waiting_consumers = Waiting}}
     end.
 
 add_waiting({Key, _} = New, Waiting) ->
@@ -2496,10 +2496,8 @@ message_size(Msg) ->
         false ->
             %% probably only hit this for testing so ok to use erts_debug
             {0, erts_debug:size(Msg)}
-
     end.
 
-
 all_nodes(#?STATE{consumers = Cons0,
                   enqueuers = Enqs0,
                   waiting_consumers = WaitingConsumers0}) ->
@@ -2577,9 +2575,10 @@ get_priority(#{priority := Priority}) ->
 get_priority(#{args := Args}) ->
     %% fallback, v3 option
     case rabbit_misc:table_lookup(Args, <<"x-priority">>) of
-        {_Key, Value} ->
+        {_Type, Value} ->
             Value;
-        _ -> 0
+        _ ->
+            0
     end;
 get_priority(_) ->
     0.
@@ -2713,7 +2712,6 @@ consumer_key_from_id(ConsumerId, {_, _, I}) ->
 consumer_key_from_id(_ConsumerId, none) ->
     error.
 
-
 consumer_cancel_info(ConsumerKey, #?STATE{consumers = Consumers}) ->
     case Consumers of
         #{ConsumerKey := #consumer{checked_out = Checked}} ->

deps/rabbit/src/rabbit_fifo_client.erl

@@ -207,33 +207,33 @@ dequeue(QueueName, ConsumerTag, Settlement,
         {ok, {dequeue, empty}, Leader} ->
             {empty, State0#state{leader = Leader}};
         {ok, {dequeue, {MsgId, {MsgHeader, Msg0}}, MsgsReady}, Leader} ->
-            {Msg, IsDelivered} = add_delivery_count_header(Msg0, MsgHeader),
+            {Msg, Redelivered} = add_delivery_count_header(Msg0, MsgHeader),
             {ok, MsgsReady,
-             {QueueName, qref(Leader), MsgId, IsDelivered, Msg},
+             {QueueName, qref(Leader), MsgId, Redelivered, Msg},
              State0#state{leader = Leader}};
         {ok, {error, _} = Err, _Leader} ->
             Err;
         Err ->
             Err
     end.
 
-add_delivery_count_header(Msg, #{delivery_count := DelCount} = Header)
+add_delivery_count_header(Msg0, #{delivery_count := DelCount} = Header)
   when is_integer(DelCount) ->
-    {case mc:is(Msg) of
-        true ->
-             %% the "delivery-count" header in the AMQP spec does not include
-             %% returns (released outcomes)
-             AmqpDelCount = DelCount - maps:get(return_count, Header, 0),
-             mc:set_annotation(delivery_count, AmqpDelCount,
-                               mc:set_annotation(<<"x-delivery-count">>,
-                                                 DelCount, Msg));
-        _ ->
-            Msg
-    end, DelCount > 0};
+    Msg = case mc:is(Msg0) of
+              true ->
+                  Msg1 = mc:set_annotation(<<"x-delivery-count">>, DelCount, Msg0),
+                  %% the "delivery-count" header in the AMQP spec does not include
+                  %% returns (released outcomes)
+                  AmqpDelCount = DelCount - maps:get(return_count, Header, 0),
+                  mc:set_annotation(delivery_count, AmqpDelCount, Msg1);
+              false ->
+                  Msg0
+          end,
+    Redelivered = DelCount > 0,
+    {Msg, Redelivered};
 add_delivery_count_header(Msg, _Header) ->
     {Msg, false}.
 
-
 %% @doc Settle a message. Permanently removes message from the queue.
 %% @param ConsumerTag the tag uniquely identifying the consumer.
 %% @param MsgIds the message ids received with the {@link rabbit_fifo:delivery/0.}

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -886,11 +886,11 @@ consume(Q, Spec, QState0) when ?amqqueue_is_quorum(Q) ->
                    _ ->
                        0
                end,
-
     Priority = case rabbit_misc:table_lookup(Args, <<"x-priority">>) of
                    {_Key, Value} ->
                        Value;
-                   _ -> 0
+                   _ ->
+                       0
                end,
     ConsumerMeta = #{ack => AckRequired,
                      prefetch => Prefetch,

deps/rabbit/test/amqp_system_SUITE.erl

@@ -69,24 +69,23 @@ init_per_group(Group, Config) ->
         dotnet -> fun build_dotnet_test_project/1;
         java   -> fun build_maven_test_project/1
     end,
-    rabbit_ct_helpers:run_setup_steps(Config1, [
-        GroupSetupStep
-      ] ++
-      rabbit_ct_broker_helpers:setup_steps() ++
-      rabbit_ct_client_helpers:setup_steps()).
+    Config2 = rabbit_ct_helpers:run_setup_steps(
+                Config1,
+                [GroupSetupStep] ++
+                rabbit_ct_broker_helpers:setup_steps() ++
+                rabbit_ct_client_helpers:setup_steps()),
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(
+           Config2, message_containers_store_amqp_v1),
+    ok = rabbit_ct_broker_helpers:enable_feature_flag(
+           Config2, quorum_queues_v4),
+    Config2.
 
 end_per_group(_, Config) ->
     rabbit_ct_helpers:run_teardown_steps(Config,
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
 
 init_per_testcase(Testcase, Config) ->
-    enable_feature_flags(Config,
-                         [
-                          message_containers_store_amqp_v1,
-                          credit_api_v2,
-                          quorum_queues_v4
-                         ]),
     rabbit_ct_helpers:testcase_started(Config, Testcase).
 
 end_per_testcase(Testcase, Config) ->
@@ -271,16 +270,6 @@ run_java_test(Config, Class) ->
       [{cd, TestProjectDir}]),
     {ok, _} = Ret.
 
-
-enable_feature_flags(Config, Flags) ->
-    [begin
-         case rabbit_ct_broker_helpers:enable_feature_flag(Config, Flag) of
-             ok -> ok;
-             _ ->
-                 throw({skip, "feature flag ~s could not be enabled"})
-         end
-     end || Flag <- Flags].
-
 declare_queue(Config, Name, Type) ->
     Ch = rabbit_ct_client_helpers:open_channel(Config),
     #'queue.declare_ok'{} =