Merge pull request #6 from coherentpath/add-support-for-item-ordering

Add support for item ordering

Author: gdwoolbert3

lib/buffer.ex

@@ -43,6 +43,10 @@ defmodule Buffer do
     * `:max_size` - The maximum size (in bytes) of the buffer before being flushed. By default,
       this limit is `:infinity`.
 
+    * `:ordering` - The order in which buffered items are returned. The options are `:fifo`
+      (first-in-first-out) and `:lifo` (last-in-first-out). Defaults to `:fifo`. Using `:lifo`
+      provides a small performance improvement by skipping the list reversal.
+
     * `:partitioner` - The method by which items are inserted into different partitions. The
       options are `:rotating` and `:random` and the former is the default.
 

lib/buffer/server.ex

@@ -12,6 +12,7 @@ defmodule Buffer.Server do
     :jitter_rate,
     :max_length,
     :max_size,
+    :ordering,
     :partition,
     :size_callback
   ]

lib/buffer/state.ex

@@ -6,6 +6,7 @@ defmodule Buffer.State do
     :flush_meta,
     :max_length,
     :max_size,
+    :ordering,
     :partition,
     :size_callback,
     :timeout,
@@ -36,6 +37,7 @@ defmodule Buffer.State do
 
   @doc false
   @spec items(t()) :: list()
+  def items(%{ordering: :lifo} = buffer), do: buffer.buffer
   def items(buffer), do: Enum.reverse(buffer.buffer)
 
   @doc false
@@ -46,12 +48,14 @@ defmodule Buffer.State do
          {:ok, size_callback} <- get_size_callback(opts),
          {:ok, max_length} <- get_max_length(opts, jitter),
          {:ok, max_size} <- get_max_size(opts, jitter),
-         {:ok, timeout} <- get_timeout(opts, jitter) do
+         {:ok, timeout} <- get_timeout(opts, jitter),
+         {:ok, ordering} <- get_ordering(opts) do
       buffer = %__MODULE__{
         flush_callback: flush_callback,
         flush_meta: Keyword.get(opts, :flush_meta),
         max_length: max_length,
         max_size: max_size,
+        ordering: ordering,
         partition: Keyword.get(opts, :partition, 0),
         size_callback: size_callback,
         timeout: timeout
@@ -103,6 +107,13 @@ defmodule Buffer.State do
     validate_limit(Keyword.get(opts, :buffer_timeout, :infinity), jitter)
   end
 
+  defp get_ordering(opts) do
+    case Keyword.get(opts, :ordering, :fifo) do
+      ordering when ordering in [:fifo, :lifo] -> {:ok, ordering}
+      _ -> {:error, :invalid_ordering}
+    end
+  end
+
   defp validate_callback(fun, arity) when is_function(fun, arity), do: {:ok, fun}
   defp validate_callback(_, _), do: {:error, :invalid_callback}
 

test/buffer_test.exs

@@ -70,6 +70,12 @@ defmodule BufferTest do
       assert start_ex_buffer(opts) == {:error, :invalid_jitter}
     end
 
+    test "will not start with an invalid ordering" do
+      opts = [ordering: :invalid]
+
+      assert start_ex_buffer(opts) == {:error, :invalid_ordering}
+    end
+
     test "will flush an Buffer on termination" do
       assert {:ok, buffer} = start_ex_buffer()
       assert seed_buffer(buffer) == :ok
@@ -177,6 +183,14 @@ defmodule BufferTest do
       assert {:ok, buffer} = start_ex_buffer()
       assert Buffer.dump(buffer, partition: -1) == {:error, :invalid_partition}
     end
+
+    test "will return items in LIFO order when set" do
+      opts = [ordering: :lifo]
+
+      assert {:ok, buffer} = start_ex_buffer(opts)
+      assert seed_buffer(buffer) == :ok
+      assert Buffer.dump(buffer) == {:ok, ["baz", "bar", "foo"]}
+    end
   end
 
   describe "flush/2" do
@@ -232,6 +246,14 @@ defmodule BufferTest do
       assert {:ok, buffer} = start_ex_buffer()
       assert Buffer.flush(buffer, partition: -1) == {:error, :invalid_partition}
     end
+
+    test "will flush items in LIFO order when set" do
+      opts = [ordering: :lifo, max_length: 3]
+
+      assert {:ok, buffer} = start_ex_buffer(opts)
+      assert seed_buffer(buffer) == :ok
+      assert_receive {^buffer, ["baz", "bar", "foo"], _}
+    end
   end
 
   describe "info/2" do