fixes #3

Author: zookzook

lib/bson/decoder.ex

@@ -8,10 +8,8 @@ defmodule BSON.Decoder do
     map
   end
 
-  def documents(binary),
-    do: documents(binary, [])
-  def documents("", acc),
-    do: Enum.reverse(acc)
+  def documents(binary), do: documents(binary, [])
+  def documents("", acc), do: Enum.reverse(acc)
   def documents(binary, acc) do
     {doc, rest} = document(binary)
     documents(rest, [doc|acc])
@@ -169,18 +167,11 @@ defmodule BSON.Decoder do
     {string, rest}
   end
 
-  defp subtype(0x00),
-    do: :generic
-  defp subtype(0x01),
-    do: :function
-  defp subtype(0x02),
-    do: :binary_old
-  defp subtype(0x03),
-    do: :uuid_old
-  defp subtype(0x04),
-    do: :uuid
-  defp subtype(0x05),
-    do: :md5
-  defp subtype(int) when is_integer(int) and int in 0x80..0xFF,
-    do: int
+  defp subtype(0x00), do: :generic
+  defp subtype(0x01), do: :function
+  defp subtype(0x02), do: :binary_old
+  defp subtype(0x03), do: :uuid_old
+  defp subtype(0x04), do: :uuid
+  defp subtype(0x05), do: :md5
+  defp subtype(int) when is_integer(int) and int in 0x80..0xFF, do: int
 end

lib/mongo/auth/cr.ex

@@ -1,19 +1,19 @@
 defmodule Mongo.Auth.CR do
   @moduledoc false
-  import Mongo.MongoDBConnection.Utils
+  alias Mongo.MongoDBConnection.Utils
 
   def auth({username, password}, s) do
-    with {:ok, message} <- command(-2, [getnonce: 1], s),
+    with {:ok, message} <- Utils.command(-2, [getnonce: 1], s),
          do: nonce(message, username, password, s)
   end
 
   defp nonce(%{"nonce" => nonce, "ok" => ok}, username, password, s)
   when ok == 1 # to support a response that returns 1 or 1.0
   do
-    digest = digest(nonce, username, password)
+    digest = Utils.digest(nonce, username, password)
     command = [authenticate: 1, user: username, nonce: nonce, key: digest]
 
-    case command(-3, command, s) do
+    case Utils.command(-3, command, s) do
       {:ok, %{"ok" => ok}} when ok == 1 ->
         :ok
       {:ok, %{"ok" => 0.0, "errmsg" => reason, "code" => code}} ->

lib/mongo/auth/scram.ex

@@ -1,9 +1,10 @@
 defmodule Mongo.Auth.SCRAM do
   @moduledoc false
   import Mongo.BinaryUtils
-  import Mongo.MongoDBConnection.Utils
   import Bitwise
 
+  alias Mongo.MongoDBConnection.Utils
+
   def auth({username, password}, s) do
     # TODO: Wrap and log error
 
@@ -13,11 +14,11 @@ defmodule Mongo.Auth.SCRAM do
     message    = [saslStart: 1, mechanism: "SCRAM-SHA-1", payload: payload]
 
     result =
-      with {:ok, %{"ok" => ok} = reply} when ok == 1 <- command(-2, message, s),
+      with {:ok, %{"ok" => ok} = reply} when ok == 1 <- Utils.command(-2, message, s),
            {message, signature} = first(reply, first_bare, username, password, nonce),
-           {:ok, %{"ok" => ok} = reply} when ok == 1 <- command(-3, message, s),
+           {:ok, %{"ok" => ok} = reply} when ok == 1 <- Utils.command(-3, message, s),
            message = second(reply, signature),
-           {:ok, %{"ok" => ok} = reply} when ok == 1 <- command(-4, message, s),
+           {:ok, %{"ok" => ok} = reply} when ok == 1 <- Utils.command(-4, message, s),
            do: final(reply)
 
     case result do
@@ -36,7 +37,7 @@ defmodule Mongo.Auth.SCRAM do
     server_nonce    = params["r"]
     salt            = params["s"] |> Base.decode64!
     iter            = params["i"] |> String.to_integer
-    pass            = digest_password(username, password)
+    pass            = Utils.digest_password(username, password)
     salted_password = hi(pass, salt, iter)
 
     <<^client_nonce::binary(24), _::binary>> = server_nonce

lib/mongo/auth/x509.ex

@@ -1,10 +1,10 @@
 defmodule Mongo.Auth.X509 do
   @moduledoc false
-  import Mongo.MongoDBConnection.Utils
+  alias Mongo.MongoDBConnection.Utils
 
   def auth({username, _password}, s) do
     cmd = [authenticate: 1, user: username, mechanism: "MONGODB-X509"]
-    with {:ok, _message} <- command(-2, cmd, s) do
+    with {:ok, _message} <- Utils.command(-2, cmd, s) do
       :ok
     else
       _error -> {:error, "X509 auth failed"}

lib/mongo/messages.ex

@@ -1,5 +1,9 @@
 defmodule Mongo.Messages do
-  @moduledoc false
+  @moduledoc """
+    This module encodes and decodes the data from and to the mongodb server.
+    We only support MongoDB >= 3.2 and use op_query with the hack collection "$cmd"
+    Other op codes are deprecated. Therefore only op_reply and op_query are supported.
+  """
 
   defmacro __using__(_opts) do
     quote do
@@ -14,21 +18,8 @@ defmodule Mongo.Messages do
   import Record
   import Mongo.BinaryUtils
 
-  @op_update        2001
-  @op_insert        2002
-  @op_query         2004
-  @op_get_more      2005
-  @op_delete        2006
-  @op_kill_cursors  2007
-
-  @update_flags [
-    upsert: 0x1,
-    multi:  0x2
-  ]
-
-  @insert_flags [
-    continue_on_error: 0x1
-  ]
+  @op_reply 1
+  @op_query 2004
 
   @query_flags [
     tailable_cursor:   0x2,
@@ -40,116 +31,71 @@ defmodule Mongo.Messages do
     partial:           0x80
   ]
 
-  @delete_flags [
-    single: 0x1
-  ]
-
   @header_size 4 * 4
 
   defrecordp :msg_header, [:length, :request_id, :response_to, :op_code]
-  defrecord  :op_update, [:coll, :flags, :query, :update]
-  defrecord  :op_insert, [:flags, :coll, :docs]
   defrecord  :op_query, [:flags, :coll, :num_skip, :num_return, :query, :select]
-  defrecord  :op_get_more, [:coll, :num_return, :cursor_id]
-  defrecord  :op_delete, [:coll, :flags, :query]
-  defrecord  :op_kill_cursors, [:cursor_ids]
   defrecord  :op_reply, [:flags, :cursor_id, :from, :num, :docs]
 
-  def encode(request_id, op) do
-    iodata = encode_op(op)
-    header = msg_header(length: IO.iodata_length(iodata) + @header_size,
-                        request_id: request_id, response_to: 0,
-                        op_code: op_to_code(op))
-
-    [encode_header(header)|iodata]
+  @doc """
+    Decodes the header from response of a request sent by the mongodb server
+  """
+  def decode_header(iolist) when is_list(iolist) do
+    case IO.iodata_length(iolist) >= @header_size do
+      true  -> iolist |> IO.iodata_to_binary() |> decode_header()
+      false -> :error
+    end
   end
+  def decode_header(<<length::int32, request_id::int32, response_to::int32, op_code::int32, rest::binary>>) do
+    header = msg_header(length: length - @header_size, request_id: request_id, response_to: response_to, op_code: op_code) ## todo don't subtract header-size here
+    {:ok, header, rest}
+  end
+  def decode_header(_binary), do: :error
 
+  @doc """
+    Decodes the response body of a request sent by the mongodb server
+  """
   def decode_response(msg_header(length: length) = header, iolist) when is_list(iolist) do
-    if IO.iodata_length(iolist) >= length,
-      do: decode_response(header, IO.iodata_to_binary(iolist)),
-    else: :error
+    case IO.iodata_length(iolist) >= length do
+      true  -> decode_response(header, IO.iodata_to_binary(iolist))
+      false -> :error
+    end
   end
-  def decode_response(msg_header(length: length, response_to: response_to), binary)  when byte_size(binary) >= length do
+  def decode_response(msg_header(length: length, response_to: response_to, op_code: op_code), binary)  when byte_size(binary) >= length do
     <<response::binary(length), rest::binary>> = binary
-    {:ok, response_to, decode_reply(response), rest}
-  end
-  def decode_response(_header, _binary) do
-    :error
+    case op_code do
+      @op_reply -> {:ok, response_to, decode_reply(response), rest}
+      _         -> :error
+    end
   end
+  def decode_response(_header, _binary), do: :error
 
-  def decode_header(iolist) when is_list(iolist) do
-    if IO.iodata_length(iolist) >= @header_size,
-      do: IO.iodata_to_binary(iolist) |> decode_header,
-    else: :error
-  end
-  def decode_header(<<length::int32, request_id::int32, response_to::int32,
-                       op_code::int32, rest::binary>>) do
-    header = msg_header(length: length-@header_size, request_id: request_id, response_to: response_to, op_code: op_code)
-    {:ok, header, rest}
-  end
-  def decode_header(_binary) do
-    :error
+  @doc """
+    Decodes a reply message from the response
+  """
+  def decode_reply(<<flags::int32, cursor_id::int64, from::int32, num::int32, rest::binary>>) do
+    op_reply(flags: flags, cursor_id: cursor_id, from: from, num: num, docs: BSON.Decoder.documents(rest))
   end
 
-  defp encode_op(op_update(coll: coll, flags: flags, query: query, update: update)) do
-    [<<0x00::int32>>,
-     coll,
-     <<0x00, blit_flags(:update, flags)::int32>>,
-     query,
-     update]
+  def encode(request_id, op_query() = op) do
+    iodata = encode_op(op)
+    header = msg_header(length: IO.iodata_length(iodata) + @header_size,  request_id: request_id, response_to: 0, op_code: @op_query)
+    [encode_header(header)|iodata]
   end
 
-  defp encode_op(op_insert(flags: flags, coll: coll, docs: docs)) do
-    [<<blit_flags(:insert, flags)::int32>>,
-     coll,
-     0x00,
-     docs]
+  defp encode_header(msg_header(length: length, request_id: request_id, response_to: response_to, op_code: op_code)) 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>>,
-     query,
+     BSON.Encoder.document(query),
      select]
   end
 
-  defp encode_op(op_get_more(coll: coll, num_return: num_return, cursor_id: cursor_id)) do
-    [<<0x00::int32>>,
-     coll,
-     <<0x00, num_return::int32, cursor_id::int64>>]
-  end
-
-  defp encode_op(op_delete(coll: coll, flags: flags, query: query)) do
-    [<<0x00::int32>>,
-     coll,
-     <<0x00, blit_flags(:delete, flags)::int32>> |
-     query]
-  end
-
-  defp encode_op(op_kill_cursors(cursor_ids: ids)) do
-    binary_ids = for id <- ids, into: "", do: <<id::int64>>
-    num = div byte_size(binary_ids), 8
-    [<<0x00::int32, num::int32>>, binary_ids]
-  end
-
-  defp op_to_code(op_update()),       do: @op_update
-  defp op_to_code(op_insert()),       do: @op_insert
-  defp op_to_code(op_query()),        do: @op_query
-  defp op_to_code(op_get_more()),     do: @op_get_more
-  defp op_to_code(op_delete()),       do: @op_delete
-  defp op_to_code(op_kill_cursors()), do: @op_kill_cursors
-
-  defp decode_reply(<<flags::int32, cursor_id::int64, from::int32, num::int32, rest::binary>>) do
-    op_reply(flags: flags, cursor_id: cursor_id, from: from, num: num, docs: rest)
-  end
-
-  defp encode_header(msg_header(length: length, request_id: request_id,
-                                response_to: response_to, op_code: op_code)) do
-    <<length::int32, request_id::int32, response_to::int32, op_code::int32>>
-  end
-
   defp blit_flags(op, flags) when is_list(flags) do
     import Bitwise
     Enum.reduce(flags, 0x0, &(flag_to_bit(op, &1) ||| &2))
@@ -158,21 +104,9 @@ defmodule Mongo.Messages do
     flags
   end
 
-  Enum.each(@update_flags, fn {flag, bit} ->
-    defp flag_to_bit(:update, unquote(flag)), do: unquote(bit)
-  end)
-
-  Enum.each(@insert_flags, fn {flag, bit} ->
-    defp flag_to_bit(:insert, unquote(flag)), do: unquote(bit)
-  end)
-
   Enum.each(@query_flags, fn {flag, bit} ->
     defp flag_to_bit(:query, unquote(flag)), do: unquote(bit)
   end)
 
-  Enum.each(@delete_flags, fn {flag, bit} ->
-    defp flag_to_bit(:delete, unquote(flag)), do: unquote(bit)
-  end)
-
   defp flag_to_bit(_op, _flag), do: 0x0
 end

lib/mongo/mongo_db_connection.ex

@@ -162,24 +162,20 @@ defmodule Mongo.MongoDBConnection do
   end
 
   @impl true
-  def handle_execute(%Mongo.Query{action: action, extra: extra} = query, params, opts, original_state) do
+  def handle_execute(%Mongo.Query{action: action} = query, params, opts, original_state) do
     tmp_state = %{original_state | database: Keyword.get(opts, :database, original_state.database)}
-    with {:ok, reply, tmp_state} <- handle_execute(action, extra, params, opts, tmp_state) do
+    with {:ok, reply, tmp_state} <- execute_action(action, params, opts, tmp_state) do
       {:ok, query, reply, Map.put(tmp_state, :database, original_state.database)}
     end
   end
 
-  defp handle_execute(:wire_version, _, _, _, state) do
+  defp execute_action(:wire_version, _, _, state) do
     {:ok, state.wire_version, state}
   end
 
-  defp handle_execute(:command, nil, [query], opts, s) do
+  defp execute_action(:command, [query], opts, state) 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))
-    |> get_response(s)
-  end
-
-  defp get_response(op, state) do
+    op = op_query(coll: Utils.namespace("$cmd", state, opts[:database]), query: query, select: "", num_skip: 0, num_return: 1, flags: flags(flags))
     with {:ok, response} <- Utils.post_request(state.request_id, op, state),
          state = %{state | request_id: state.request_id + 1},
          do: {:ok, response, state}

lib/mongo/query.ex

@@ -1,27 +1,22 @@
 defmodule Mongo.Query do
-  @moduledoc false
+  @moduledoc """
+    This is the query implementation for the Query Protocoll
 
-  defstruct action: nil, extra: nil, encoded?: false
+    The action attribute describes as atom the desired action. There are currently two
+    * :command
+    * :wire_version
+      
+    Encoding and decoding does not take place at this point, but is directly performed
+    into the functions of Mongo.MongoDBConnection.Utils.
+  """
+  defstruct action: nil
 end
 
 defimpl DBConnection.Query, for: Mongo.Query do
-  import Mongo.Messages, only: [op_reply: 1, op_reply: 2]
-
   def parse(query, _opts), do: query
   def describe(query, _opts), do: query
-
-  def encode(query, params, _opts) do
-    if query.encoded? do
-      params
-    else
-      Enum.map(params, fn
-        nil -> ""
-        doc -> BSON.Encoder.document(doc)
-      end)
-    end
-  end
-
+  def encode(query, params, _opts), do: params
   def decode(_query, :ok, _opts), do: :ok
   def decode(_query, wire_version, _opts) when is_integer(wire_version), do: wire_version
-  def decode(_query, op_reply(docs: docs) = reply, _opts), do: op_reply(reply, docs: BSON.Decoder.documents(docs))
+  def decode(_query, reply, _opts), do: reply
 end