first test on api

Author: zookzook

lib/mongo.ex

@@ -296,7 +296,7 @@ defmodule Mongo do
 
   end
 
-  def bulk_write(topology_pid, coll, %UnorderedBulk{inserts: inserts, updates: updates, deletes: deletes}, opts) do
+  def bulk_write(topology_pid, %UnorderedBulk{coll: coll, inserts: inserts, updates: updates, deletes: deletes}, opts \\ []) do
 
     {_ids, inserts} = assign_ids(inserts)
 
@@ -307,47 +307,57 @@ defmodule Mongo do
                     } |> filter_nils()
 
     insert_cmd = [
-            insert: coll, ## todo
+            insert: coll,
             documents: inserts,
-            ordered: Keyword.get(opts, :ordered),
-            writeConcern: write_concern,
-            bypassDocumentValidation: Keyword.get(opts, :bypass_document_validation)
+            writeConcern: write_concern
           ] |> filter_nils()
 
-    deletes = Enum.map(deletes, fn filter -> %{
-                               q: filter,
-                               limit: 1,
-                               collation: Keyword.get(opts, :collation)
-                             } |> filter_nils() end)
+    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,   ## todo
+            delete: coll,
             deletes: deletes,
             ordered: Keyword.get(opts, :ordered),
             writeConcern: write_concern
           ] |> filter_nils()
 
-    updates = Enum.map(updates, fn {filter, update} -> %{
+    updates = Enum.map(updates, fn {filter, update, update_opts} -> %{
                q: filter,
                u: update,
-               upsert: Keyword.get(opts, :upsert),
-               multi: false,
-               collation: Keyword.get(opts, :collation),
-               arrayFilters: Keyword.get(opts, :filters)
+               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),
+            # ordered: Keyword.get(opts, :ordered),
             writeConcern: write_concern,
-            bypassDocumentValidation: Keyword.get(opts, :bypass_document_validation)
+            # 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"]}
+    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 """
@@ -626,15 +636,14 @@ 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, cmds, opts \\ []) when is_list(cmds) do
+  def direct_command(conn, cmd, opts) do
     action = %Query{action: :command}
 
-    with {:ok, _cmd, doc} <- DBConnection.execute(conn, action, cmds, defaults(opts)),
+    with {:ok, _cmd, doc} <- DBConnection.execute(conn, action, [cmd], 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/mongo_db_connection.ex

@@ -95,17 +95,15 @@ defmodule Mongo.MongoDBConnection do
     end
   end
 
-  defp tcp_connect(opts, state) do
+  defp tcp_connect(opts, s) do
+    host      = (opts[:hostname] || "localhost") |> to_charlist
+    port      = opts[:port] || 27017
+    sock_opts = [:binary, active: false, packet: :raw, nodelay: true]
+    ++ (opts[:socket_options] || [])
 
-    {host, port} = Utils.hostname_port(opts)
-    sock_opts   = [:binary, active: false, packet: :raw, nodelay: true] ++ (opts[:socket_options] || [])
+    s = Map.put(s, :host, "#{host}:#{port}")
 
-    s = case host do
-      {:local, socket} -> Map.put(state, :host, socket)
-      hostname         -> Map.put(state, :host, "#{hostname}:#{port}")
-    end
-
-    case :gen_tcp.connect(host, port, sock_opts, state.connect_timeout) do
+    case :gen_tcp.connect(host, port, sock_opts, s.connect_timeout) do
       {:ok, socket} ->
         # A suitable :buffer is only set if :recbuf is included in
         # :socket_options.
@@ -115,7 +113,7 @@ defmodule Mongo.MongoDBConnection do
 
         {:ok, %{s | connection: {:gen_tcp, socket}}}
 
-      {:error, reason} -> {:error, Mongo.Error.exception(tag: :tcp, action: "connect", reason: reason, host: state.host)}
+      {:error, reason} -> {:error, Mongo.Error.exception(tag: :tcp, action: "connect", reason: reason, host: s.host)}
     end
   end
 
@@ -244,31 +242,24 @@ defmodule Mongo.MongoDBConnection do
   defp execute_action(:wire_version, _, _, state) do
     {:ok, state.wire_version, state}
   end
-  defp execute_action(:command, cmds, opts, %{wire_version: version} = state) when version >= 6 do
+  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])]]
 
-    mode    = update_read_preferences(opts[:slave_ok])
     timeout = Keyword.get(opts, :max_time, 0)
 
-    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)
+    # 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
 
     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)
@@ -300,7 +291,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)))
 
-    {payload_0, payload_1}
+    op_msg(flags: 0, sections: [payload_0, payload_1])
   end
 
   def update_read_preferences(true), do: "primaryPreferred"

lib/mongo/unordered_bulk.ex

@@ -5,26 +5,48 @@ defmodule Mongo.UnorderedBulk do
 
   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 inserts: [], updates: [], deletes: [], opts: []
+  defstruct coll: nil, inserts: [], updates: [], deletes: [], opts: []
+
+  def new(coll) do
+    %UnorderedBulk{coll: coll}
+  end
 
-  def new() do
-    %UnorderedBulk{}
+  def insert_one(%UnorderedBulk{inserts: rest} = bulk, doc) do
+    %UnorderedBulk{bulk | inserts: [doc | rest] }
   end
 
-  def insert_one(%UnorderedBulk{inserts: rest} = b, doc) do
-    %UnorderedBulk{b | inserts: [doc | rest] }
+  def delete_one(%UnorderedBulk{deletes: rest} = bulk, doc, collaction \\ nil) do
+    %UnorderedBulk{bulk | deletes: [{doc, collaction, false} | rest] }
   end
 
-  def delete_one(%UnorderedBulk{deletes: rest} = b, doc) do
-    %UnorderedBulk{b | deletes: [doc | rest] }
+  def delete_many(%UnorderedBulk{deletes: rest} = bulk, doc, collaction \\ nil) do
+    %UnorderedBulk{bulk | deletes: [{doc, collaction, true} | rest] }
   end
 
-  def update_one(%UnorderedBulk{updates: rest} = b, filter, update) do
-    %UnorderedBulk{b | updates: [{filter, update} | rest] }
+  def update_one(%UnorderedBulk{updates: rest} = bulk, filter, update, opts \\ []) do
+    %UnorderedBulk{bulk | updates: [{filter, update, opts} | rest] }
   end
 
+  def test() do
+
+    "bulk"
+    |> new()
+    |> insert_one(%{name: "Greta"})
+    |> insert_one(%{name: "Tom"})
+    |> insert_one(%{name: "Waldo"})
+    |> update_one(%{name: "Greta"}, %{"$set": %{kind: "dog"}})
+    |> update_one(%{name: "Tom"}, %{"$set": %{kind: "dog"}})
+    |> update_one(%{name: "Waldo"}, %{"$set": %{kind: "dog"}})
+    |> delete_one(%{kind: "dog"})
+    |> delete_one(%{kind: "dog"})
+    |> delete_one(%{kind: "dog"})
+
+
+  end
 end