update batch upsert function

diff --git a/lib/cluster_kv/db.ex b/lib/cluster_kv/db.ex
index 721414f2899b8183452c5ffd97f8583b22ca46c1..734e70ce86e2e07bde847a0599a34f9615128b19 100644 (file)
--- a/lib/cluster_kv/db.ex
+++ b/lib/cluster_kv/db.ex
@@ -73,7 +73,7 @@ defmodule ClusterKV.DB do
           chunk :: integer(),
           fun :: update_function()
         ) :: :ok
-  def batch(name, batch, chunk \\ 10, fun \\ &{elem(&1, 0), &2}) do
+  def batch(name, batch, chunk \\ 10, fun \\ &{elem(&1, 0), Enum.uniq([elem(&1, 1) | &2])}) do
     :poolboy.transaction(name, fn w ->
       GenServer.cast(w, {:batch, batch, chunk, fun})
     end)

diff --git a/lib/cluster_kv/ring.ex b/lib/cluster_kv/ring.ex
index 52267c13e5dd06fa98bf3d8d0cdf49eed37a69f9..198959e5cb2367bdf2da14f6fd8e414c537ef59f 100644 (file)
--- a/lib/cluster_kv/ring.ex
+++ b/lib/cluster_kv/ring.ex
@@ -478,7 +478,7 @@ defmodule ClusterKV.Ring do
     Logger.debug("Sending batches to: #{inspect(Map.keys(batches))}")
 
     Enum.each(batches, fn {node, batch} ->
-      Logger.debug("Sending #{length(batch)} reccords to #{inspect(node)}")
+      Logger.debug("Sending #{length(batch)} records to #{inspect(node)}")
       send({name, node}, {:handle_batch, batch})
     end)
   end
@@ -598,7 +598,13 @@ defmodule ClusterKV.Ring do
     end)
   end
 
-  @spec send_sync(name :: module(), nodes :: [node()], key :: String.t(), value :: any()) :: :ok
+  @spec send_update(
+          name :: module(),
+          nodes :: [node()],
+          key :: String.t(),
+          value :: any(),
+          fun :: any()
+        ) :: :ok
   defp send_update(name, nodes, key, value, fun) do
     Enum.each(nodes, fn n ->
       send({name, n}, {:update, key, value, fun})