replaced op_update by update command

Author: Michael Maier

lib/mongo.ex

@@ -695,7 +695,7 @@ defmodule Mongo do
   # This is the implementation of the delete command for
   # delete_one and delete_many
   #
-  defp delete_documents(topology_pid, coll, filter, limit, opts \\ [])  do
+  defp delete_documents(topology_pid, coll, filter, limit, opts)  do
 
     # see https://docs.mongodb.com/manual/reference/command/delete/#dbcmd.delete
     write_concern = %{
@@ -749,20 +749,7 @@ defmodule Mongo do
   @spec replace_one(GenServer.server, collection, BSON.document, BSON.document, Keyword.t) :: result(Mongo.UpdateResult.t)
   def replace_one(topology_pid, coll, filter, replacement, opts \\ []) do
     _ = modifier_docs(replacement, :replace)
-
-    params = [filter, replacement]
-    query = %Query{action: :replace_one, extra: coll}
-    with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, reply} <- DBConnection.execute(conn, query, params, defaults(opts)),
-         :ok <- maybe_failure(reply),
-         {:ok, doc} <- get_last_error(reply) do
-      case doc do
-        %{"n" => 1, "upserted" => upserted_id} ->
-          {:ok, %Mongo.UpdateResult{matched_count: 0, modified_count: 1, upserted_id: upserted_id}}
-        %{"n" => n} ->
-          {:ok, %Mongo.UpdateResult{matched_count: n, modified_count: n}}
-      end
-    end
+    update_documents(topology_pid, coll, filter, replacement, false, opts)
   end
 
   @doc """
@@ -795,20 +782,7 @@ defmodule Mongo do
   @spec update_one(GenServer.server, collection, BSON.document, BSON.document, Keyword.t) :: result(Mongo.UpdateResult.t)
   def update_one(topology_pid, coll, filter, update, opts \\ []) do
     _ = modifier_docs(update, :update)
-
-    params = [filter, update]
-    query = %Query{action: :update_one, extra: coll}
-    with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, reply} <- DBConnection.execute(conn, query, params, defaults(opts)),
-         :ok <- maybe_failure(reply),
-         {:ok, doc} <- get_last_error(reply) do
-      case doc do
-        %{"n" => 1, "upserted" => upserted_id} ->
-          {:ok, %Mongo.UpdateResult{matched_count: 0, modified_count: 1, upserted_id: upserted_id}}
-        %{"n" => n} ->
-          {:ok, %Mongo.UpdateResult{matched_count: n, modified_count: n}}
-      end
-    end
+    update_documents(topology_pid, coll, filter, update, false, opts)
   end
 
   @doc """
@@ -834,18 +808,46 @@ defmodule Mongo do
   @spec update_many(GenServer.server, collection, BSON.document, BSON.document, Keyword.t) :: result(Mongo.UpdateResult.t)
   def update_many(topology_pid, coll, filter, update, opts \\ []) do
     _ = modifier_docs(update, :update)
+    update_documents(topology_pid, coll, filter, update, true, opts)
+  end
+
+  defp update_documents(topology_pid, coll, filter, update, multi, opts) do
+
+    # see https://docs.mongodb.com/manual/reference/command/update/#update-command-output
+    # validation of the update document
+    write_concern = %{
+                      w: Keyword.get(opts, :w),
+                      j: Keyword.get(opts, :j),
+                      wtimeout: Keyword.get(opts, :wtimeout)
+                    } |> filter_nils()
+
+    update = %{
+               q: filter,
+               u: update,
+               upsert: Keyword.get(opts, :upsert),
+               multi: multi,
+               collation: Keyword.get(opts, :ordered),
+               arrayFilters: Keyword.get(opts, :filters)
+             } |> filter_nils()
+
+    query = [
+              update: coll,
+              updates: [update],
+              ordered: Keyword.get(opts, :ordered),
+              writeConcern: write_concern,
+              bypassDocumentValidation: Keyword.get(opts, :bypass_document_validation)
+            ] |> filter_nils()
 
-    params = [filter, update]
-    query = %Query{action: :update_many, extra: coll}
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, reply} <- DBConnection.execute(conn, query, params, defaults(opts)),
-         :ok <- maybe_failure(reply),
-         {:ok, doc} <- get_last_error(reply) do
+         {:ok, doc} <- direct_command(conn, query, opts) do
       case doc do
-        %{"n" => 1, "upserted" => upserted_id} ->
-          {:ok, %Mongo.UpdateResult{matched_count: 0, modified_count: 1, upserted_id: upserted_id}}
-        %{"n" => n} ->
-          {:ok, %Mongo.UpdateResult{matched_count: n, modified_count: n}}
+        # todo: better handling off the result
+        %{"writeErrors" => _} -> {:error, %Mongo.WriteError{n: doc["n"], ok: doc["ok"], write_errors: doc["writeErrors"]}}
+        %{"n" => n, "nModified" => n_modified} ->
+          case Map.get(write_concern, :w) do
+            0 -> {:ok, %Mongo.UpdateResult{acknowledged: false}}
+            _ -> {:ok, %Mongo.UpdateResult{matched_count: n, modified_count: n_modified}}
+          end
       end
     end
   end

lib/mongo/mongo_db_connection.ex

@@ -9,7 +9,6 @@ defmodule Mongo.MongoDBConnection do
 
   @timeout        5000
   @find_one_flags ~w(slave_ok exhaust partial)a
-  @update_flags   ~w(upsert)a
   @write_concern  ~w(w j wtimeout)a
 
   def connect(opts) do
@@ -144,27 +143,6 @@ defmodule Mongo.MongoDBConnection do
     {:ok, state.wire_version, state}
   end
 
-  defp handle_execute(:replace_one, coll, [query, replacement], opts, s) do
-    flags  = flags(Keyword.take(opts, @update_flags))
-    op     = op_update(coll: Utils.namespace(coll, s, opts[:database]), query: query, update: replacement,
-                       flags: flags)
-    message_gle(-15, op, opts, s)
-  end
-
-  defp handle_execute(:update_one, coll, [query, update], opts, s) do
-    flags  = flags(Keyword.take(opts, @update_flags))
-    op     = op_update(coll: Utils.namespace(coll, s, opts[:database]), query: query, update: update,
-                       flags: flags)
-    message_gle(-16, op, opts, s)
-  end
-
-  defp handle_execute(:update_many, coll, [query, update], opts, s) do
-    flags  = [:multi | flags(Keyword.take(opts, @update_flags))]
-    op     = op_update(coll: Utils.namespace(coll, s, opts[:database]), query: query, update: update,
-                       flags: flags)
-    message_gle(-17, op, opts, s)
-  end
-
   defp handle_execute(:command, nil, [query], opts, s) do
     flags = Keyword.take(opts, @find_one_flags)
     op_query(coll: Utils.namespace("$cmd", s, opts[:database]), query: query, select: "", num_skip: 0, num_return: 1, flags: flags(flags))
@@ -184,22 +162,6 @@ defmodule Mongo.MongoDBConnection do
     end)
   end
 
-  defp message_gle(id, op, opts, s) do
-    write_concern = Keyword.take(opts, @write_concern) |> Map.new
-    write_concern = Map.merge(s.write_concern, write_concern)
-
-    if write_concern.w == 0 do
-      with :ok <- Utils.send(id, op, s), do: {:ok, :ok, s}
-    else
-      command = BSON.Encoder.document([{:getLastError, 1}|Map.to_list(write_concern)])
-      gle_op = op_query(coll: Utils.namespace("$cmd", s, opts[:database]), query: command,
-                        select: "", num_skip: 0, num_return: -1, flags: [])
-
-      ops = [{id, op}, {s.request_id, gle_op}]
-      get_response(ops, s)
-    end
-  end
-
   def ping(%{wire_version: wire_version} = state) do
     with {:ok, %{wire_version: ^wire_version}} <- wire_version(state), do: {:ok, state}
   end

lib/mongo/results.ex

@@ -54,10 +54,11 @@ defmodule Mongo.UpdateResult do
   """
 
   @type t :: %__MODULE__{
+    acknowledged: boolean,
     matched_count: non_neg_integer,
     modified_count: non_neg_integer,
     upserted_id: nil | BSON.ObjectId.t
   }
 
-  defstruct [:matched_count, :modified_count, :upserted_id]
+  defstruct [acknowledged: true, matched_count: 0, modified_count: 0, upserted_id: 0]
 end