QQ: v9 - module setup.

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -12,15 +12,15 @@
 -compile(inline_list_funcs).
 -compile(inline).
 -compile({no_auto_import, [apply/3]}).
--dialyzer({nowarn_function, convert_v7_to_v8/2}).
+-dialyzer({nowarn_function, convert_v8_to_v9/2}).
 -dialyzer(no_improper_lists).
 
 -include("rabbit_queue_type.hrl").
 -include("rabbit_fifo.hrl").
 
 -include_lib("kernel/include/logger.hrl").
 
--define(STATE, ?MODULE).
+-define(STATE, rabbit_fifo).
 -define(DLX, rabbit_fifo_dlx).
 
 -define(CONSUMER_PID(Pid), #consumer{cfg = #consumer_cfg{pid = Pid}}).
@@ -903,8 +903,8 @@ finalize_ra_seq({SeqAcc, Start, End}) ->
       State :: state(),
       OldMeta :: ra_snapshot:meta(),
       OldState :: state().
-snapshot_installed(_Meta, #?MODULE{cfg = #cfg{},
-                                   consumers = Consumers} = State,
+snapshot_installed(_Meta, #?STATE{cfg = #cfg{},
+                                  consumers = Consumers} = State,
                    _OldMeta, _OldState) ->
     %% here we need to redliver all pending consumer messages
     %% to local consumers
@@ -931,8 +931,8 @@ snapshot_installed(_Meta, #?MODULE{cfg = #cfg{},
     delivery_effects(SendAcc, State) ++
     credit_reply_resend_effect(State).
 
-credit_reply_resend_effect(#?MODULE{waiting_consumers = Waiting,
-                                    consumers = Consumers} = State) ->
+credit_reply_resend_effect(#?STATE{waiting_consumers = Waiting,
+                                   consumers = Consumers} = State) ->
     Available = messages_ready(State),
     maps:fold(
       fun(ConsumerKey,
@@ -966,69 +966,9 @@ credit_reply_resend_effect(#?MODULE{waiting_consumers = Waiting,
               Acc
       end, [], maps:merge(Consumers, maps:from_list(Waiting))).
 
-v7_to_v8_consumer(Con, Timeout) ->
-                     V7Cfg = element(#consumer.cfg, Con),
-                     Status0 = element(#consumer.status, Con),
-                     Ch0 = element(#consumer.checked_out, Con),
-                     Ch = maps:map(fun (_, M) -> ?C_MSG(Timeout, M) end, Ch0),
-                     Cfg = #consumer_cfg{meta = element(#consumer_cfg.meta, V7Cfg),
-                                         pid = element(#consumer_cfg.pid, V7Cfg),
-                                         tag = element(#consumer_cfg.tag, V7Cfg),
-                                         credit_mode = element(#consumer_cfg.credit_mode, V7Cfg),
-                                         lifetime = element(#consumer_cfg.lifetime, V7Cfg),
-                                         priority = element(#consumer_cfg.priority, V7Cfg),
-                                         timeout = ?DEFAULT_CONSUMER_TIMEOUT_MS
-                                        },
-                     Status = case Status0 of
-                                  suspected_down ->
-                                      {suspected_down, up};
-                                  _ ->
-                                      Status0
-                              end,
-                     #consumer{cfg = Cfg,
-                               status = Status,
-                               next_msg_id = element(#consumer.next_msg_id, Con),
-                               checked_out = Ch,
-                               credit = element(#consumer.credit, Con),
-                               delivery_count = element(#consumer.delivery_count, Con)
-                              }.
-
-convert_v7_to_v8(#{system_time := Ts} = _Meta, StateV7) ->
-    %% the structure is intact for now
-    Cons0 = element(#?STATE.consumers, StateV7),
-    Waiting0 = element(#?STATE.waiting_consumers, StateV7),
-    Timeout = Ts + ?DEFAULT_CONSUMER_TIMEOUT_MS,
-    Cons = maps:map(
-             fun (_CKey, Con) ->
-                     v7_to_v8_consumer(Con, Timeout)
-             end, Cons0),
-    Waiting = lists:map(fun({Cid, Con}) ->
-                                {Cid, v7_to_v8_consumer(Con, Timeout)}
-                        end, Waiting0),
-
-    Msgs = element(#?STATE.messages, StateV7),
-    Cfg = element(#?STATE.cfg, StateV7),
-    {Hi, No} = rabbit_fifo_q:to_queues(Msgs),
-    Pq0 = queue:fold(fun (I, Acc) ->
-                             rabbit_fifo_pq:in(9, I, Acc)
-                     end, rabbit_fifo_pq:new(), Hi),
-    Pq = queue:fold(fun (I, Acc) ->
-                            rabbit_fifo_pq:in(?DEFAULT_PRIORITY, I, Acc)
-                    end, Pq0, No),
-    Dlx0 = element(#?STATE.dlx, StateV7),
-    Dlx = Dlx0#?DLX{unused = ?NIL},
-    StateV8 = StateV7,
-    StateV8#?STATE{cfg = Cfg#cfg{consumer_disconnected_timeout = 60_000,
-                                 delayed_retry = disabled},
-                   reclaimable_bytes = 0,
-                   messages = Pq,
-                   consumers = Cons,
-                   waiting_consumers = Waiting,
-                   next_consumer_timeout = Timeout,
-                   last_command_time = Ts,
-                   dlx = Dlx,
-                   delayed = #delayed{}
-                  }.
+convert_v8_to_v9(#{} = _Meta, StateV8) ->
+    State = StateV8,
+    State.
 
 purge_node(Meta, Node, State, Effects) ->
     lists:foldl(fun(Pid, {S0, E0}) ->
@@ -1252,7 +1192,7 @@ get_checked_out(CKey, From, To, #?STATE{consumers = Consumers}) ->
     end.
 
 -spec version() -> pos_integer().
-version() -> 8.
+version() -> 9.
 
 which_module(0) -> rabbit_fifo_v0;
 which_module(1) -> rabbit_fifo_v1;
@@ -1262,7 +1202,8 @@ which_module(4) -> rabbit_fifo_v7;
 which_module(5) -> rabbit_fifo_v7;
 which_module(6) -> rabbit_fifo_v7;
 which_module(7) -> rabbit_fifo_v7;
-which_module(8) -> ?MODULE.
+which_module(8) -> rabbit_fifo_v8;
+which_module(9) -> ?MODULE.
 
 -define(AUX, aux_v4).
 
@@ -1729,10 +1670,7 @@ query_notify_decorators_info(#?STATE{consumers = Consumers} = State) ->
 
 -spec usage(atom()) -> float().
 usage(Name) when is_atom(Name) ->
-    case ets:lookup(rabbit_fifo_usage, Name) of
-        [] -> 0.0;
-        [{_, Use}] -> Use
-    end.
+    ets:lookup_element(rabbit_fifo_usage, Name, 2, 0.0).
 
 %%% Internal
 
@@ -3582,7 +3520,9 @@ convert(Meta, 6, To, State) ->
     %% no conversion needed, this version only includes a logic change
     convert(Meta, 7, To, State);
 convert(Meta, 7, To, State) ->
-    convert(Meta, 8, To, convert_v7_to_v8(Meta, State)).
+    convert(Meta, 8, To, rabbit_fifo_v8:convert_v7_to_v8(Meta, State));
+convert(Meta, 8, To, State) ->
+    convert(Meta, 9, To, convert_v8_to_v9(Meta, State)).
 
 smallest_raft_index(#?STATE{messages = Messages,
                             returns = Returns,
@@ -4266,7 +4206,7 @@ consumer_info(ConsumerKey,
                         delivery_count = DeliveryCount,
                         next_msg_id = NextMsgId},
               State) ->
-    #?MODULE{cfg = #cfg{consumer_strategy = Strat}} = State,
+    #?STATE{cfg = #cfg{consumer_strategy = Strat}} = State,
     #{next_msg_id => NextMsgId,
       credit => Credit,
       key => ConsumerKey,
@@ -4277,8 +4217,8 @@ consumer_info(ConsumerKey,
 
 handle_waiting_timedout_consumers(Meta, Key, MsgIds, State0) ->
     case State0 of
-        #?MODULE{cfg = #cfg{consumer_strategy = single_active},
-                 waiting_consumers = Waiting0} ->
+        #?STATE{cfg = #cfg{consumer_strategy = single_active},
+                waiting_consumers = Waiting0} ->
             %% check if consumer key relates to any of the waiting
             %% consumers
             case find_waiting_consumer(Key, Waiting0) of
@@ -4292,12 +4232,12 @@ handle_waiting_timedout_consumers(Meta, Key, MsgIds, State0) ->
                             Waiting = add_waiting({CKey, WC}, Waiting0),
                             {State, Effects} =
                             activate_next_consumer(
-                              State0#?MODULE{waiting_consumers = Waiting}, []),
+                              State0#?STATE{waiting_consumers = Waiting}, []),
                             checkout(Meta, State0, State, Effects);
                         Rem ->
                             WC = WC0#consumer{timed_out_msg_ids = Rem},
                             Waiting = add_waiting({CKey, WC}, Waiting0),
-                            {State0#?MODULE{waiting_consumers  = Waiting}, ok}
+                            {State0#?STATE{waiting_consumers  = Waiting}, ok}
                     end;
                 _ ->
                     {State0, ok}