1 files changed, 119 insertions, 0 deletions
diff --git a/deps/rabbit/src/rabbit_fifo_index.erl b/deps/rabbit/src/rabbit_fifo_index.erl
new file mode 100644
index 0000000000..14ac89faff
--- /dev/null
+++ b/deps/rabbit/src/rabbit_fifo_index.erl
@@ -0,0 +1,119 @@
+-module(rabbit_fifo_index).
+
+-export([
+         empty/0,
+         exists/2,
+         append/2,
+         delete/2,
+         size/1,
+         smallest/1,
+         map/2
+        ]).
+
+-compile({no_auto_import, [size/1]}).
+
+%% the empty atom is a lot smaller (4 bytes) than e.g. `undefined` (13 bytes).
+%% This matters as the data map gets persisted as part of the snapshot
+-define(NIL, '').
+
+-record(?MODULE, {data = #{} :: #{integer() => ?NIL},
+                  smallest :: undefined | non_neg_integer(),
+                  largest :: undefined | non_neg_integer()
+                 }).
+
+
+-opaque state() :: #?MODULE{}.
+
+-export_type([state/0]).
+
+-spec empty() -> state().
+empty() ->
+    #?MODULE{}.
+
+-spec exists(integer(), state()) -> boolean().
+exists(Key, #?MODULE{data = Data}) ->
+    maps:is_key(Key, Data).
+
+% only integer keys are supported
+-spec append(integer(), state()) -> state().
+append(Key,
+       #?MODULE{data = Data,
+                smallest = Smallest,
+                largest = Largest} = State)
+  when Key > Largest orelse Largest =:= undefined ->
+    State#?MODULE{data = maps:put(Key, ?NIL, Data),
+                  smallest = ra_lib:default(Smallest, Key),
+                  largest = Key}.
+
+-spec delete(Index :: integer(), state()) -> state().
+delete(Smallest, #?MODULE{data = Data0,
+                          largest = Largest,
+                          smallest = Smallest} = State) ->
+    Data = maps:remove(Smallest, Data0),
+    case find_next(Smallest + 1, Largest, Data) of
+        undefined ->
+            State#?MODULE{data = Data,
+                          smallest = undefined,
+                          largest = undefined};
+        Next ->
+            State#?MODULE{data = Data, smallest = Next}
+    end;
+delete(Key, #?MODULE{data = Data} = State) ->
+    State#?MODULE{data = maps:remove(Key, Data)}.
+
+-spec size(state()) -> non_neg_integer().
+size(#?MODULE{data = Data}) ->
+    maps:size(Data).
+
+-spec smallest(state()) -> undefined | integer().
+smallest(#?MODULE{smallest = Smallest}) ->
+    Smallest.
+
+
+-spec map(fun(), state()) -> state().
+map(F, #?MODULE{data = Data} = State) ->
+    State#?MODULE{data = maps:map(F, Data)}.
+
+
+%% internal
+
+find_next(Next, Last, _Map) when Next > Last ->
+    undefined;
+find_next(Next, Last, Map) ->
+    case Map of
+        #{Next := _} ->
+            Next;
+        _ ->
+            % in degenerate cases the range here could be very large
+            % and hence this could be very slow
+            % the typical case should ideally be better
+            % assuming fifo-ish deletion of entries
+            find_next(Next+1, Last, Map)
+    end.
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
+append_test() ->
+    S0 = empty(),
+    false = exists(99, S0),
+    undefined = smallest(S0),
+    0 = size(S0),
+    S1 = append(1, S0),
+    false = exists(99, S1),
+    true = exists(1, S1),
+    1 = size(S1),
+    1 = smallest(S1),
+    S2 = append(2, S1),
+    true = exists(2, S2),
+    2 = size(S2),
+    1 = smallest(S2),
+    S3 = delete(1, S2),
+    2 = smallest(S3),
+    1 = size(S3),
+    S5 = delete(2, S3),
+    undefined = smallest(S5),
+    0 = size(S0),
+    ok.
+
+-endif.