preparations for #12

Author: zookzook

lib/mongo.ex

@@ -624,7 +624,7 @@ defmodule Mongo do
       wtimeout: Keyword.get(opts, :wtimeout)
     } |> filter_nils()
 
-    query = [
+    cmd = [
       insert: coll,
       documents: [doc],
       ordered: Keyword.get(opts, :ordered),
@@ -633,7 +633,7 @@ defmodule Mongo do
     ] |> filter_nils()
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc} <- direct_command(conn, query, opts) do
+         {:ok, doc} <- direct_command(conn, cmd, opts) do
       case doc do
         %{"writeErrors" => _} -> {:error, %Mongo.WriteError{n: doc["n"], ok: doc["ok"], write_errors: doc["writeErrors"]}}
         _ ->
@@ -869,7 +869,7 @@ defmodule Mongo do
                arrayFilters: Keyword.get(opts, :filters)
              } |> filter_nils()
 
-    query = [
+    cmd = [
               update: coll,
               updates: [update],
               ordered: Keyword.get(opts, :ordered),
@@ -878,7 +878,7 @@ defmodule Mongo do
             ] |> filter_nils()
 
     with {:ok, conn, _, _} <- select_server(topology_pid, :write, opts),
-         {:ok, doc}        <- direct_command(conn, query, opts) do
+         {:ok, doc}        <- direct_command(conn, cmd, opts) do
 
       case doc do
 

lib/mongo/grid_fs/upload_stream.ex

@@ -137,11 +137,17 @@ defmodule Mongo.GridFs.UploadStream do
     #
     defp insert_one_file_document(topology_pid, collection, file_id, length, chunk_size, filename, metadata, opts) do
       doc = %{_id: file_id, length: length, filename: filename, chunkSize: chunk_size, uploadDate: now(), metadata: metadata}
+            |> filter_nils()
+
       {:ok, _} = Mongo.insert_one(topology_pid, collection, doc, opts)
     end
 
     defp now(), do: DateTime.from_unix!(:os.system_time(), :native)
 
+    defp filter_nils(map) when is_map(map) do
+      Enum.reject(map, fn {_key, value} -> is_nil(value) end)
+      |> Enum.into(%{})
+    end
   end
 
 end

lib/mongo/messages.ex

@@ -125,32 +125,36 @@ defmodule Mongo.Messages do
     <<length::int32, request_id::int32, response_to::int32, op_code::int32>>
   end
 
+  defp encode_op(op_query(flags: flags, coll: coll, num_skip: num_skip,
+    num_return: num_return, query: query, select: select)) do
+    [<<blit_flags(:query, flags)::int32>>,
+      coll,
+      <<0x00, num_skip::int32, num_return::int32>>,
+      BSON.Encoder.document(query),
+      select]
+  end
   defp encode_op(op_msg(flags: flags, sections: sections)) do
     # todo: flags encoding
-    [<<0::int32>>, encode_sections(sections)]
+    [<<0::int32>> | encode_sections(sections)]
   end
+
   defp encode_sections(sections) do
     Enum.map(sections, fn section -> encode_section(section) end)
   end
 
   defp encode_section(section(payload_type: t, payload: payload)) do
-    [<<t::int8>>, encode_payload(payload)]
+    [<<t::int8>> | encode_payload(payload)]
   end
   defp encode_payload(payload(doc: doc, sequence: nil)) do
     BSON.Encoder.document(doc)
   end
-  defp encode_payload(payload(doc: nil, sequence: sequence(size: size, identifier: identifier, docs: docs))) do
-    [<<size::int32>>, identifier, <<0x00>>, BSON.Encoder.encode(docs)]
+  defp encode_payload(payload(doc: nil, sequence: sequence(identifier: identifier, docs: docs))) do
+    iodata = [identifier, <<0x00>> | Enum.map(docs, fn doc -> BSON.Encoder.encode(doc) end)]
+    size   = IO.iodata_length(iodata) + 4
+    [<<size::int32>> | iodata]
   end
 
-  defp encode_op(op_query(flags: flags, coll: coll, num_skip: num_skip,
-                          num_return: num_return, query: query, select: select)) do
-    [<<blit_flags(:query, flags)::int32>>,
-     coll,
-     <<0x00, num_skip::int32, num_return::int32>>,
-     BSON.Encoder.document(query),
-     select]
-  end
+
 
   defp blit_flags(op, flags) when is_list(flags) do
     import Bitwise

lib/mongo/mongo_db_connection.ex

@@ -248,7 +248,12 @@ defmodule Mongo.MongoDBConnection do
                   "$readPreference": [mode: update_read_preferences(opts[:slave_ok])]]
 
     timeout = Keyword.get(opts, :max_time, 0)
-    op      = op_msg(flags: 0, sections: [section(payload_type: 0, payload: payload(doc: cmd))])
+
+    # 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
@@ -271,6 +276,24 @@ defmodule Mongo.MongoDBConnection do
     {:disconnect, exception, state}
   end
 
+  defp pulling_out?(cmd, key) do
+    case Keyword.has_key?(cmd, key) do
+      true  -> key
+      false -> nil
+    end
+  end
+
+  defp pulling_out(cmd, key) when is_atom(key) do
+
+    docs = Keyword.get(cmd, key)
+    cmd  = Keyword.delete(cmd, key)
+
+    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])
+  end
+
   def update_read_preferences(true), do: "primaryPreferred"
   def update_read_preferences(false), do: "primary"
   def update_read_preferences(nil), do: "primary"