added support for slaveOk, added test case for sharded configuration.

Author: zookzook

lib/mongo/change_stream.ex

@@ -12,9 +12,8 @@ defmodule Mongo.ChangeStream do
     ## check, if retryable reads are enabled
     opts = Mongo.retryable_reads(opts)
 
-    with new_cmd = Mongo.ReadPreference.add_read_preference(cmd, opts),
-         {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
-         {:ok, %{"ok" => ok} = doc} when ok == 1 <- Mongo.exec_command_session(session, new_cmd, opts) do
+    with {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
+         {:ok, %{"ok" => ok} = doc} when ok == 1 <- Mongo.exec_command_session(session, cmd, opts) do
       %Mongo.ChangeStream{
         topology_pid: topology_pid,
         session: session,
@@ -70,9 +69,8 @@ defmodule Mongo.ChangeStream do
 
     def aggregate(topology_pid, cmd, fun, opts) do
 
-      with new_cmd = Mongo.ReadPreference.add_read_preference(cmd, opts),
-           {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
-           {:ok, %{"ok" => ok} = doc} when ok == 1 <- Mongo.exec_command_session(session, new_cmd, opts)  do
+      with {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
+           {:ok, %{"ok" => ok} = doc} when ok == 1 <- Mongo.exec_command_session(session, cmd, opts)  do
 
         aggregate(topology_pid, session, doc, cmd, fun)
       else

lib/mongo/read_preference.ex

@@ -53,44 +53,41 @@ defmodule Mongo.ReadPreference do
   Add read preference to the cmd
   """
   def add_read_preference(cmd, opts) do
+    case Keyword.get(opts, :read_preference) do
+      nil  -> cmd
+      pref -> cmd ++ ["$readPreference": pref]
+    end
+  end
+
+  @doc """
+  From the specs:
 
-    read_preference = opts
-                      |> Keyword.get(:read_preference)
-                      |> Mongo.ReadPreference.primary()
-                      |> transform()
+  Use of slaveOk
 
-    cmd ++ ["$readPreference": read_preference]
-  end
+  There are two usages of slaveOK:
 
-  defp transform(%{:mode => :primary}) do
+  * A driver query parameter that predated read preference modes and tag set lists.
+  * A wire protocol flag on OP_QUERY operations
+
+  """
+  def slave_ok(%{:mode => :primary}) do
     %{:mode => :primary}
   end
-  defp transform(config) do
+  def slave_ok(config) do
 
     mode = case config[:mode] do
       :primary_preferred   -> :primaryPreferred
       :secondary_preferred -> :secondaryPreferred
-      other -> other
-    end
-
-    max_staleness_seconds = case config[:max_staleness_ms] do
-      i when is_integer(i) -> div(i, 1000)
-      nil                  -> nil
+      other                -> other
     end
 
-    [mode: mode,
-      tag_sets: config[:tag_sets],
-      maxStalenessSeconds: max_staleness_seconds,
-      hedge: config[:hedge]]
-    |> filter_nils()
-
+    filter_nils([mode: mode, tag_sets: config[:tag_sets]])
   end
-
   ##
   # Therefore, when sending queries to a mongos, the following rules apply:
   #
   # For mode 'primary', drivers MUST NOT set the slaveOK wire protocol flag and MUST NOT use $readPreference
-  def mongos(%{mode: :primay}) do
+  def mongos(%{mode: :primary}) do
     nil
   end
   # For mode 'secondary', drivers MUST set the slaveOK wire protocol flag and MUST also use $readPreference
@@ -112,6 +109,28 @@ defmodule Mongo.ReadPreference do
     transform(config)
   end
 
+  defp transform(%{:mode => :primary}) do
+    %{:mode => :primary}
+  end
+  defp transform(config) do
+
+    mode = case config[:mode] do
+      :primary_preferred   -> :primaryPreferred
+      :secondary_preferred -> :secondaryPreferred
+      other -> other
+    end
+
+    max_staleness_seconds = case config[:max_staleness_ms] do
+      i when is_integer(i) -> div(i, 1000)
+      nil                  -> nil
+    end
 
+    [mode: mode,
+      tag_sets: config[:tag_sets],
+      maxStalenessSeconds: max_staleness_seconds,
+      hedge: config[:hedge]]
+    |> filter_nils()
+
+  end
 
 end

lib/mongo/session.ex

@@ -99,10 +99,11 @@ defmodule Mongo.Session do
   import Keywords
   import Mongo.WriteConcern
 
-  alias Mongo.Session.ServerSession
+  alias BSON.Timestamp
+  alias Mongo.ReadPreference
   alias Mongo.Session
+  alias Mongo.Session.ServerSession
   alias Mongo.Topology
-  alias BSON.Timestamp
 
   @type t :: pid()
 
@@ -430,7 +431,10 @@ defmodule Mongo.Session do
       _     -> [lsid: %{id: id}, readConcern: read_concern(data, Keyword.get(cmd, :readConcern))]
     end
 
-    cmd = Keyword.merge(cmd, options) |> filter_nils()
+    cmd = cmd
+          |> Keyword.merge(options)
+          |> ReadPreference.add_read_preference(opts)
+          |> filter_nils()
 
     {:keep_state_and_data, {:ok, conn, cmd}}
   end

lib/mongo/stream.ex

@@ -14,8 +14,7 @@ defmodule Mongo.Stream do
     ## check, if retryable reads are enabled
     opts = Mongo.retryable_reads(opts)
 
-    with cmd = Mongo.ReadPreference.add_read_preference(cmd, opts),
-         {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
+    with {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
          {:ok,
            %{"ok" => ok,
              "cursor" => %{

lib/mongo/topology_description.ex

@@ -96,7 +96,7 @@ defmodule Mongo.TopologyDescription do
       :unknown -> {[], nil}
       :single  -> {topology.servers, nil}
       :sharded -> {mongos_servers(topology), ReadPreference.mongos(read_preference)}
-      _        -> {select_replica_set_server(topology, read_preference.mode, read_preference), nil}
+      _        -> {select_replica_set_server(topology, read_preference.mode, read_preference), ReadPreference.slave_ok(read_preference)}
     end
 
     opts = case read_prefs do

test/mongo/topology_description_test.exs

@@ -19,6 +19,34 @@ defmodule Mongo.TopologyDescriptionTest do
     assert {:ok, {^single_server, _}} = TopologyDescription.select_servers(single(), :read, opts)
   end
 
+
+  test "shared server selection" do
+    sharded_server = "localhost:27017"
+
+    assert {:ok, {^sharded_server, []}} = TopologyDescription.select_servers(sharded(), :write, [])
+
+    opts = [
+      read_preference: ReadPreference.primary(%{mode: :primary})
+    ]
+    assert {:ok, {^sharded_server, []}} = TopologyDescription.select_servers(sharded(), :read, opts)
+    opts = [
+      read_preference: ReadPreference.primary(%{mode: :secondary})
+    ]
+    assert {:ok, {^sharded_server, [{:read_preference, [mode: :secondary, tag_sets: [], maxStalenessSeconds: 0]}]}} = TopologyDescription.select_servers(sharded(), :read, opts)
+    opts = [
+      read_preference: ReadPreference.primary(%{mode: :primary_preferred})
+    ]
+    assert {:ok, {^sharded_server, [{:read_preference, [mode: :primaryPreferred, tag_sets: [], maxStalenessSeconds: 0]}]}} = TopologyDescription.select_servers(sharded(), :read, opts)
+    opts = [
+      read_preference: ReadPreference.primary(%{mode: :secondary_preferred})
+    ]
+    assert {:ok, {^sharded_server, [{:read_preference, [mode: :secondaryPreferred, tag_sets: [], maxStalenessSeconds: 0]}]}} = TopologyDescription.select_servers(sharded(), :read, opts)
+    opts = [
+      read_preference: ReadPreference.primary(%{mode: :nearest})
+    ]
+    assert {:ok, {^sharded_server, [{:read_preference, [mode: :nearest, tag_sets: [], maxStalenessSeconds: 0]}]}} = TopologyDescription.select_servers(sharded(), :read, opts)
+  end
+
   test "replica set server selection" do
     all_hosts = ["localhost:27018", "localhost:27019", "localhost:27020"]
     master = "localhost:27018"

test/support/topology_test_data.ex

@@ -31,6 +31,38 @@ defmodule Mongo.TopologyTestData do
     }
   }
 
+  def sharded(), do: %{
+    set_name: nil,
+    type: :sharded,
+    compatibility_error: nil,
+    compatible: true,
+    local_threshold_ms: 15,
+    max_election_id: nil,
+    max_set_version: nil,
+    servers: %{
+      "localhost:27017" => %{
+        address: "localhost:27017",
+        arbiters: [],
+        election_id: nil,
+        error: nil,
+        hosts: [],
+        last_update_time: nil,
+        last_write_date: nil,
+        max_wire_version: 8,
+        me: nil,
+        min_wire_version: 0,
+        op_time: nil,
+        passives: [],
+        primary: nil,
+        round_trip_time: 44,
+        set_name: nil,
+        set_version: nil,
+        tag_set: %{},
+        type: :mongos
+      }
+    }
+  }
+
   def repl_set_with_master, do: %{
     compatibility_error: nil,
     compatible: true,