refactor code

Author: zookzook

lib/mongo.ex

@@ -296,70 +296,6 @@ defmodule Mongo do
 
   end
 
-  def bulk_write(topology_pid, %UnorderedBulk{coll: coll, inserts: inserts, updates: updates, deletes: deletes}, opts \\ []) do
-
-    {_ids, inserts} = assign_ids(inserts)
-
-    write_concern = %{
-                      w: Keyword.get(opts, :w),
-                      j: Keyword.get(opts, :j),
-                      wtimeout: Keyword.get(opts, :wtimeout)
-                    } |> filter_nils()
-
-    insert_cmd = [
-            insert: coll,
-            documents: inserts,
-            writeConcern: write_concern
-          ] |> filter_nils()
-
-    deletes = Enum.map(deletes,
-                       fn {filter, collaction, many} ->
-                          limit = case many do
-                            true  -> 0
-                            false -> 1
-                          end
-                          %{
-                             q: filter,
-                             limit: limit,
-                             collation: collaction
-                           } |> filter_nils()
-                       end)
-
-    delete_cmd = [
-            delete: coll,
-            deletes: deletes,
-            ordered: Keyword.get(opts, :ordered),
-            writeConcern: write_concern
-          ] |> filter_nils()
-
-    updates = Enum.map(updates, fn {filter, update, update_opts} -> %{
-               q: filter,
-               u: update,
-               upsert: Keyword.get(update_opts, :upsert),
-               multi: Keyword.get(update_opts, :multi) || false,
-               collation: Keyword.get(update_opts, :collation),
-               arrayFilters: Keyword.get(update_opts, :filters)
-             } |> filter_nils() end)
-
-    update_cmd = [
-            update: coll,
-            updates: updates,
-            # ordered: Keyword.get(opts, :ordered),
-            writeConcern: write_concern,
-            # bypassDocumentValidation: Keyword.get(opts, :bypass_document_validation)
-          ] |> filter_nils()
-
-    cmds = [insert_cmd, update_cmd, delete_cmd]
-
-    IO.puts inspect cmds
-
-    with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts) do
-      cmds
-      |> Enum.map(fn cmd -> direct_command(conn, cmd, opts) end)
-      |> Enum.map(fn {:ok, doc} -> {:ok, doc} end)
-    end
-  end
-
   @doc """
   Finds a document and replaces it.
 

lib/mongo/bulk_write.ex

@@ -0,0 +1,123 @@
+defmodule Mongo.BulkWrite do
+  @moduledoc """
+
+  todo
+
+  Ist immer für eine Collections
+
+   The maxWriteBatchSize limit of a database, which indicates the maximum number of write operations permitted in a write batch, raises from 1,000 to 100,000.
+
+  """
+
+  import Mongo.Utils
+  alias Mongo.UnorderedBulk
+
+  @doc """
+  Unordered bulk write operations:
+  Executes first insert commands, then all update commands and after that all delete commands are executed. If a group (inserts, updates or deletes) exceeds the limit
+  maxWriteBatchSize it will be split into chunks. Everything is done in memory, so this use case is limited by memory. A better approach seems to use streaming bulk writes.
+  """
+  def bulk_write(topology_pid, %UnorderedBulk{} = bulk, opts \\ []) do
+
+    write_concern = %{
+                      w: Keyword.get(opts, :w),
+                      j: Keyword.get(opts, :j),
+                      wtimeout: Keyword.get(opts, :wtimeout)
+                    } |> filter_nils()
+
+    with {:ok, conn, _, _} <- Mongo.select_server(topology_pid, :write, opts),
+         inserts <- conn |> run_commands(get_insert_cmds(bulk, write_concern), opts) |> collect(:inserts),
+         updates <- conn |> run_commands(get_update_cmds(bulk, write_concern, opts), opts) |> collect(:updates),
+         deletes <- conn |> run_commands(get_delete_cmds(bulk, write_concern, opts), opts) |> collect(:deletes) do
+      inserts ++ updates ++ deletes
+    end
+  end
+
+  def collect(doc, :inserts) do
+
+  end
+
+  def collect(doc, :updates) do
+
+  end
+
+  def collect(doc, :deletes) do
+
+  end
+
+  defp run_commands(conn, cmds, opts) do
+
+    IO.puts "Running cmsd #{inspect cmds}"
+
+    cmds
+    |> Enum.map(fn cmd -> Mongo.direct_command(conn, cmd, opts) end)
+    |> Enum.map(fn {:ok, doc} -> {:ok, doc} end)
+  end
+
+  def get_insert_cmds(%UnorderedBulk{coll: coll, inserts: all_inserts}, write_concern) do
+
+    max_batch_size = 10 ## only for test maxWriteBatchSize
+
+    {_ids, all_inserts} = assign_ids(all_inserts)
+
+    all_inserts
+    |> Enum.chunk_every(max_batch_size)
+    |> Enum.map(fn inserts -> get_insert_cmd(coll, inserts, write_concern) end)
+
+  end
+
+  defp get_insert_cmd(coll, inserts, write_concern) do
+    filter_nils([insert: coll, documents: inserts, writeConcern: write_concern])
+  end
+
+  defp get_delete_cmds(%UnorderedBulk{coll: coll, deletes: all_deletes}, write_concern, opts) do
+
+    max_batch_size = 10 ## only for test maxWriteBatchSize
+    all_deletes
+    |> Enum.chunk_every(max_batch_size)
+    |> Enum.map(fn deletes -> get_delete_cmd(coll, deletes, write_concern, opts) end)
+
+  end
+
+  defp get_delete_cmd(coll, deletes, write_concern, opts ) do
+    filter_nils([delete: coll,
+                 deletes: Enum.map(deletes, fn delete -> get_delete_doc(delete) end),
+                 ordered: Keyword.get(opts, :ordered),
+                 writeConcern: write_concern])
+  end
+  defp get_delete_doc({filter, collaction, limit}) do
+    %{q: filter, limit: limit, collation: collaction} |> filter_nils()
+  end
+
+  defp get_update_cmds(%UnorderedBulk{coll: coll, updates: all_updates}, write_concern, opts) do
+
+    max_batch_size = 10 ## only for test maxWriteBatchSize
+    all_updates
+    |> Enum.chunk_every(max_batch_size)
+    |> Enum.map(fn updates -> get_update_cmd(coll, updates, write_concern, opts) end)
+
+  end
+
+  defp get_update_cmd(coll, updates, write_concern, opts) do
+    [ update: coll,
+      updates: Enum.map(updates, fn update -> get_update_doc(update) end),
+      ordered: Keyword.get(opts, :ordered),
+      writeConcern: write_concern,
+      bypassDocumentValidation: Keyword.get(opts, :bypass_document_validation)
+    ] |> filter_nils()
+  end
+
+  defp get_update_doc({filter, update, update_opts}) do
+    [ q: filter,
+      u: update,
+      upsert: Keyword.get(update_opts, :upsert),
+      multi: Keyword.get(update_opts, :multi) || false,
+      collation: Keyword.get(update_opts, :collation),
+      arrayFilters: Keyword.get(update_opts, :filters)
+    ] |> filter_nils()
+  end
+  defp get_update_doc(_other) do
+    []
+  end
+
+end

lib/mongo/unordered_bulk.ex

@@ -1,17 +1,13 @@
 defmodule Mongo.UnorderedBulk do
   @moduledoc """
 
-  todo
-
-  Ist immer für eine Collections
-
    The maxWriteBatchSize limit of a database, which indicates the maximum number of write operations permitted in a write batch, raises from 1,000 to 100,000.
 
   """
 
   alias Mongo.UnorderedBulk
 
-  defstruct coll: nil, inserts: [], updates: [], deletes: [], opts: []
+  defstruct coll: nil, inserts: [], updates: [], deletes: []
 
   def new(coll) do
     %UnorderedBulk{coll: coll}
@@ -22,20 +18,20 @@ defmodule Mongo.UnorderedBulk do
   end
 
   def delete_one(%UnorderedBulk{deletes: rest} = bulk, doc, collaction \\ nil) do
-    %UnorderedBulk{bulk | deletes: [{doc, collaction, false} | rest] }
+    %UnorderedBulk{bulk | deletes: [{doc, collaction, 1} | rest] }
   end
 
   def delete_many(%UnorderedBulk{deletes: rest} = bulk, doc, collaction \\ nil) do
-    %UnorderedBulk{bulk | deletes: [{doc, collaction, true} | rest] }
+    %UnorderedBulk{bulk | deletes: [{doc, collaction, 0} | rest] }
   end
 
   def update_one(%UnorderedBulk{updates: rest} = bulk, filter, update, opts \\ []) do
     %UnorderedBulk{bulk | updates: [{filter, update, opts} | rest] }
   end
 
-  def test() do
+  def test(top) do
 
-    "bulk"
+    bulk = "bulk"
     |> new()
     |> insert_one(%{name: "Greta"})
     |> insert_one(%{name: "Tom"})
@@ -47,6 +43,8 @@ defmodule Mongo.UnorderedBulk do
     |> delete_one(%{kind: "dog"})
     |> delete_one(%{kind: "dog"})
 
+    result = Mongo.BulkWrite.bulk_write(top, bulk)
 
+    IO.puts inspect result
   end
 end

lib/utils.ex

@@ -0,0 +1,43 @@
+defmodule Mongo.Utils do
+
+
+  def filter_nils(keyword) when is_list(keyword) do
+    Enum.reject(keyword, fn {_key, value} -> is_nil(value) end)
+  end
+
+  def filter_nils(map) when is_map(map) do
+    Enum.reject(map, fn {_key, value} -> is_nil(value) end)
+    |> Enum.into(%{})
+  end
+
+
+  def assign_ids(list) when is_list(list) do
+    Enum.map(list, &assign_id/1)
+    |> Enum.unzip
+  end
+
+  defp assign_id(%{_id: id} = map) when id != nil,  do: {id, map}
+  defp assign_id(%{"_id" => id} = map) when id != nil, do: {id, map}
+  defp assign_id([{_, _} | _] = keyword) do
+    case Keyword.take(keyword, [:_id, "_id"]) do
+      [{_key, id} | _] when id != nil -> {id, keyword}
+      [] -> add_id(keyword)
+    end
+  end
+
+  defp assign_id(map) when is_map(map) do
+    map |> Map.to_list |> add_id
+  end
+
+  ##
+  # Inserts an ID to the document. A distinction is made as to whether binaries or atoms are used as keys.
+  #
+  defp add_id(doc) do
+    id = Mongo.IdServer.new
+    {id, add_id(doc, id)}
+  end
+  defp add_id([{key, _}|_] = list, id) when is_atom(key), do: [{:_id, id}|list]
+  defp add_id([{key, _}|_] = list, id) when is_binary(key), do: [{"_id", id}|list]
+  defp add_id([], id), do: [{"_id", id}]
+
+end