brainstorming for bulk api

Author: zookzook

lib/mongo.ex

@@ -53,6 +53,7 @@ defmodule Mongo do
   alias Mongo.TopologyDescription
   alias Mongo.Topology
   alias Mongo.UrlParser
+  alias Mongo.UnorderedBulk
 
   @timeout 15000 # 5000
 
@@ -288,6 +289,59 @@ defmodule Mongo do
 
   end
 
+  def bulk_write(topology_pid, coll, %UnorderedBulk{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, ## todo
+            documents: inserts,
+            ordered: Keyword.get(opts, :ordered),
+            writeConcern: write_concern,
+            bypassDocumentValidation: Keyword.get(opts, :bypass_document_validation)
+          ] |> filter_nils()
+
+    deletes = Enum.map(deletes, fn filter -> %{
+                               q: filter,
+                               limit: 1,
+                               collation: Keyword.get(opts, :collation)
+                             } |> filter_nils() end)
+
+    delete_cmd = [
+            delete: coll,   ## todo
+            deletes: deletes,
+            ordered: Keyword.get(opts, :ordered),
+            writeConcern: write_concern
+          ] |> filter_nils()
+
+    updates = Enum.map(updates, fn {filter, update} -> %{
+               q: filter,
+               u: update,
+               upsert: Keyword.get(opts, :upsert),
+               multi: false,
+               collation: Keyword.get(opts, :collation),
+               arrayFilters: Keyword.get(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, delete_cmd, update_cmd]
+    with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
+         {:ok, doc} <- direct_command(conn, cmds, opts), do: {:ok, doc["value"]}
+  end
 
   @doc """
   Finds a document and replaces it.
@@ -565,14 +619,15 @@ defmodule Mongo do
 
   @doc false
   @spec direct_command(pid, BSON.document, Keyword.t) :: {:ok, BSON.document | nil} | {:error, Mongo.Error.t}
-  def direct_command(conn, cmd, opts \\ []) do
+  def direct_command(conn, cmds, opts \\ []) when is_list(cmds) do
     action = %Query{action: :command}
 
-    with {:ok, _cmd, doc} <- DBConnection.execute(conn, action, [cmd], defaults(opts)),
+    with {:ok, _cmd, doc} <- DBConnection.execute(conn, action, cmds, defaults(opts)),
          {:ok, doc} <- check_for_error(doc) do
       {:ok, doc}
     end
   end
+  def direct_command(conn, cmd, opts), do: direct_command(conn, [cmd], opts)
 
   defp check_for_error(%{"ok" => ok} = response) when ok == 1, do: {:ok, response}
   defp check_for_error(%{"code" => code, "errmsg" => msg}), do: {:error, Mongo.Error.exception(message: msg, code: code)}

lib/mongo/messages.ex

@@ -154,8 +154,6 @@ defmodule Mongo.Messages do
     [<<size::int32>> | iodata]
   end
 
-
-
   defp blit_flags(op, flags) when is_list(flags) do
     import Bitwise
     Enum.reduce(flags, 0x0, &(flag_to_bit(op, &1) ||| &2))

lib/mongo/mongo_db_connection.ex

@@ -242,24 +242,31 @@ defmodule Mongo.MongoDBConnection do
   defp execute_action(:wire_version, _, _, state) do
     {:ok, state.wire_version, state}
   end
-  defp execute_action(:command, [cmd], opts, %{wire_version: version} = state) when version >= 6 do
-
-    cmd = cmd ++ ["$db": opts[:database] || state.database,
-                  "$readPreference": [mode: update_read_preferences(opts[:slave_ok])]]
+  defp execute_action(:command, cmds, opts, %{wire_version: version} = state) when version >= 6 do
 
+    mode    = update_read_preferences(opts[:slave_ok])
     timeout = Keyword.get(opts, :max_time, 0)
 
-    # MongoDB 3.6 only allows certain command arguments to be provided this way. These are:
-    op = case pulling_out?(cmd, :documents) || pulling_out?(cmd, :updates) || pulling_out?(cmd, :deletes) do
-      nil -> op_msg(flags: 0, sections: [section(payload_type: 0, payload: payload(doc: cmd))])
-      key -> pulling_out(cmd, key)
-    end
+    sections = cmds
+    |> Enum.map(fn cmd -> cmd ++ ["$db": opts[:database] || state.database, "$readPreference": [mode: mode]] end)
+    |> Enum.recude([], fn cmd, acc ->
+          # MongoDB 3.6 only allows certain command arguments to be provided this way. These are:
+          case pulling_out?(cmd, :documents) || pulling_out?(cmd, :updates) || pulling_out?(cmd, :deletes) do
+          nil -> [section(payload_type: 0, payload: payload(doc: cmd)) | acc]
+          key ->
+                 {payload_0, payload_1} = pulling_out(cmd, key)
+                 [payload_0 | [payload_1 | acc]]
+        end
+      end)
+    op = op_msg(flags: 0, sections: sections)
 
     with {:ok, doc} <- Utils.post_request(op, state.request_id, state, timeout),
          state = %{state | request_id: state.request_id + 1} do
       {:ok, doc, state}
+
     end
   end
+
   defp execute_action(:command, [cmd], opts, state) do
 
     timeout = Keyword.get(opts, :max_time, 0)
@@ -291,7 +298,7 @@ defmodule Mongo.MongoDBConnection do
     payload_0 = section(payload_type: 0, payload: payload(doc: cmd))
     payload_1 = section(payload_type: 1, payload: payload(sequence: sequence(identifier: to_string(key), docs: docs)))
 
-    op_msg(flags: 0, sections: [payload_0, payload_1])
+    {payload_0, payload_1}
   end
 
   def update_read_preferences(true), do: "primaryPreferred"

lib/mongo/unordered_bulk.ex

@@ -0,0 +1,30 @@
+defmodule Mongo.UnorderedBulk do
+  @moduledoc """
+
+  todo
+
+  Ist immer für eine Collections
+
+  """
+
+  alias Mongo.UnorderedBulk
+
+  defstruct inserts: [], updates: [], deletes: [], opts: []
+
+  def new() do
+    %UnorderedBulk{}
+  end
+
+  def insert_one(%UnorderedBulk{inserts: rest} = b, doc) do
+    %UnorderedBulk{b | inserts: [doc | rest] }
+  end
+
+  def delete_one(%UnorderedBulk{deletes: rest} = b, doc) do
+    %UnorderedBulk{b | deletes: [doc | rest] }
+  end
+
+  def update_one(%UnorderedBulk{updates: rest} = b, filter, update) do
+    %UnorderedBulk{b | updates: [{filter, update} | rest] }
+  end
+
+end