Update ReadBuffer to add update listener, and specify behavior for in…

…serting data

.travis.yml

@@ -0,0 +1,8 @@
+language: elixir
+elixir:
+  - 1.1.0
+  - 1.2.0
+otp_release:
+  - 18.0
+script:
+  - "MIX_ENV=test mix do deps.get, compile, coveralls.travis"

VERSION

@@ -1 +1 @@
-0.1.1
+0.1.2

lib/buffer/read.ex

@@ -1,31 +1,30 @@
 defmodule Buffer.Read do
   use GenServer
 
-  @default_interval 1000
-
   defmacro __using__(_) do
     quote do
       import unquote(__MODULE__), only: :macros
     end
   end
 
   defmacro buffer(opts) do
-    interval = :proplists.get_value(:interval, opts, @default_interval)
-    read = :proplists.get_value(:read, opts)
     quote do
       def worker do
         import Supervisor.Spec
-        buffer = %{
+        state = %{
           name: __MODULE__,
-          interval: unquote(interval),
-          read: unquote(read)
+          interval: unquote(opts[:interval]),
+          read: unquote(opts[:read]),
+          on_element_updated: unquote(opts[:on_element_updated]),
+          behavior: unquote(opts[:behavior])
         }
-        worker(unquote(__MODULE__), [buffer], id: __MODULE__)
+        worker(unquote(__MODULE__), [state], id: __MODULE__)
       end
 
-      def read(key), do: unquote(__MODULE__).read(__MODULE__, key)
-
+      def get(key), do: unquote(__MODULE__).get(__MODULE__, key)
+      def select(match_spec), do: unquote(__MODULE__).select(__MODULE__, match_spec)
       def sync(), do: unquote(__MODULE__).sync(__MODULE__)
+      def dump_table(), do: unquote(__MODULE__).dump_table(__MODULE__)
     end
   end
 
@@ -43,13 +42,23 @@ defmodule Buffer.Read do
     {:ok, state}
   end
 
-  def read(name, key) do
+  def get(name, key) do
     case :ets.lookup(name, key) do
       [{_, value}] -> value
       _ -> nil
     end
   end
 
+  def dump_table(name), do: :ets.tab2list(name)
+
+  def select(name, match_spec) do
+    :ets.select(name, match_spec)
+  end
+
+  def delete(name, key) do
+    :ets.delete(name, key)
+  end
+
   def handle_call(:sync, _, state) do
     read(state)
     {:reply, :ok, state}
@@ -63,5 +72,28 @@ defmodule Buffer.Read do
     {:noreply, state}
   end
 
-  defp read(state), do: :ets.insert(state.name, state.read.())
+  defp read(state) do
+    elements = state.read.()
+
+    if state.behavior == :delete do
+      match_spec = [{{:"$1", :_}, [], [:"$1"]}]
+      old_ids = select(state.name, match_spec)
+      new_ids = Enum.map(elements, fn({id, _}) -> id end)
+      for id <- (old_ids -- new_ids), do: delete(state.name, id)
+    end
+
+    if state.on_element_updated != nil do
+      updated_ids = Enum.reduce(elements, [], fn({id, element}, acc) ->
+        if element != get(state.name, id) do
+          [id | acc]
+        else
+          acc
+        end
+      end)
+
+      unless updated_ids == [], do: state.on_element_updated.(updated_ids)
+    end
+
+    :ets.insert(state.name, elements)
+  end
 end

lib/buffer/write/count.ex

@@ -1,26 +1,22 @@
 defmodule Buffer.Write.Count do
   use GenServer
 
-  @default_interval 1000
-
   defmacro __using__(_) do
     quote do
       import unquote(__MODULE__), only: :macros
     end
   end
 
   defmacro buffer(opts) do
-    interval = :proplists.get_value(:interval, opts, @default_interval)
-    write = :proplists.get_value(:write, opts)
     quote do
       def worker do
         import Supervisor.Spec
-        buffer = %{
+        state = %{
           name: __MODULE__,
-          interval: unquote(interval),
-          write: unquote(write)
+          interval: unquote(opts[:interval]),
+          write: unquote(opts[:write])
         }
-        worker(unquote(__MODULE__), [buffer], id: __MODULE__)
+        worker(unquote(__MODULE__), [state], id: __MODULE__)
       end
 
       def incr(key), do: unquote(__MODULE__).incr(__MODULE__, key, 1)

lib/buffer/write/key_list.ex

@@ -1,31 +1,26 @@
 defmodule Buffer.Write.KeyList do
   use GenServer
 
-  @default_interval 1000
-
   defmacro __using__(_) do
     quote do
       import unquote(__MODULE__), only: :macros
     end
   end
 
   defmacro buffer(opts) do
-    interval = :proplists.get_value(:interval, opts, @default_interval)
-    write = :proplists.get_value(:write, opts)
-    limit = :proplists.get_value(:limit, opts, nil)
     quote do
       def worker do
         import Supervisor.Spec
-        buffer = %{
+        state = %{
           name: __MODULE__,
-          interval: unquote(interval),
-          write: unquote(write),
-          limit: unquote(limit)
+          interval: unquote(opts[:interval]),
+          write: unquote(opts[:write]),
+          limit: unquote(opts[:limit])
         }
-        worker(unquote(__MODULE__), [buffer], id: __MODULE__)
+        worker(unquote(__MODULE__), [state], id: __MODULE__)
       end
 
-      def add(key, element), do: unquote(__MODULE__).add(__MODULE__, key, element, unquote(limit))
+      def add(key, element), do: unquote(__MODULE__).add(__MODULE__, key, element, unquote(opts[:limit]))
 
       def dump_table(), do: unquote(__MODULE__).dump_table(__MODULE__)
       def reset(), do: unquote(__MODULE__).reset(__MODULE__)

mix.exs

@@ -8,9 +8,10 @@ defmodule Buffer.Mixfile do
       app: :buffer,
       version: @version,
       elixir: "~> 1.2",
+      test_coverage: [tool: ExCoveralls],
       build_embedded: Mix.env == :prod,
       start_permanent: Mix.env == :prod,
-      deps: [],
+      deps: [{:excoveralls, "~> 0.5.1", only: :test}],
       package: package,
       name: "buffer",
       source_url: "https://github.com/adrienmo/buffer",

test/buffer_test.exs

@@ -49,9 +49,61 @@ defmodule BufferTest do
     assert([{_, [key2: 5050, key1: 100]}] = result)
   end
 
-  test "0100# Read, Read" do
+  test "0100# Read, Get" do
     BufferRead.sync()
-    assert(BufferRead.read(:key1) == "value1")
+    assert(BufferRead.get(:key1) == "value1")
+  end
+
+  test "0101# Read, Select" do
+    BufferRead.sync()
+    match_spec1 = get_match_spec("fn(x = {_, %{field1: 5}}) -> x end")
+    match_spec2 = get_match_spec("fn(x = {_, %{field2: 4}}) -> x end")
+
+    field1_5 = BufferRead.select(match_spec1)
+    field2_4 = BufferRead.select(match_spec2)
+
+    assert(field1_5[:key3] != nil)
+    assert(field2_4[:key3] != nil)
+    assert(field2_4[:key4] != nil)
+  end
+
+  test "0102# Read, Update" do
+    BufferKeyListResult.add(:key1, :value1)
+    BufferKeyListResult.add(:key2, :value2)
+
+    BufferReadUpdate.sync()
+    BufferKeyListResult.reset()
+
+    BufferKeyListResult.add(:key1, :value2)
+    BufferKeyListResult.add(:key2, :value2)
+
+    BufferReadUpdate.sync()
+    result = BufferKeyListResult.dump_table()
+    assert(result[BufferReadUpdate] == [:key1])
+  end
+
+  test "0103# Read, Delete" do
+    BufferKeyListResult.add(:key1, :value1)
+    BufferKeyListResult.add(:key2, :value2)
+
+    BufferReadDefaultBehavior.sync()
+    BufferReadDeleteBehavior.sync()
+
+    BufferKeyListResult.reset()
+    BufferKeyListResult.add(:key3, :value3)
+
+    BufferReadDefaultBehavior.sync()
+    BufferReadDeleteBehavior.sync()
+
+    assert(length(BufferReadDefaultBehavior.dump_table()) == 3)
+    assert(length(BufferReadDeleteBehavior.dump_table()) == 1)
+  end
+
+  defp get_match_spec(fun_string) do
+    fun_string
+    |> Code.eval_string()
+    |> elem(0)
+    |> :ets.fun2ms()
   end
 end
 
@@ -67,7 +119,10 @@ defmodule TestSupervisor do
       BufferKeyListLimit.worker,
       BufferKeyListInterval.worker,
       BufferCount.worker,
-      BufferRead.worker
+      BufferRead.worker,
+      BufferReadUpdate.worker,
+      BufferReadDefaultBehavior.worker,
+      BufferReadDeleteBehavior.worker
     ]
     supervise(children, strategy: :one_for_one, max_restarts: 1, max_seconds: 1)
   end
@@ -105,8 +160,41 @@ end
 
 defmodule BufferRead do
   use Buffer.Read
-  buffer interval: 1000, read: &read/0
+  buffer read: &read/0
+  def read() do
+    [
+      {:key1, "value1"},
+      {:key2, "value2"},
+      {:key3, %{field1: 5, field2: 4}},
+      {:key4, %{field1: 4, field2: 4}}
+    ]
+  end
+end
+
+defmodule BufferReadUpdate do
+  use Buffer.Read
+  buffer read: &read/0, on_element_updated: &update/1
+  def read() do
+    BufferKeyListResult.dump_table()
+  end
+
+  def update(x) do
+    BufferKeyListResult.add(__MODULE__, x)
+  end
+end
+
+defmodule BufferReadDefaultBehavior do
+  use Buffer.Read
+  buffer read: &read/0
+  def read() do
+    BufferKeyListResult.dump_table()
+  end
+end
+
+defmodule BufferReadDeleteBehavior do
+  use Buffer.Read
+  buffer read: &read/0, behavior: :delete
   def read() do
-    [{:key1, "value1"}, {:key2, "value2"}]
+    BufferKeyListResult.dump_table()
   end
 end