read preferences are only sent, when the server is a mongos.

Author: zookzook

lib/mongo.ex

@@ -392,13 +392,11 @@ defmodule Mongo do
   """
   def issue_command(topology_pid, cmd, :read, opts) do
 
-    new_cmd = ReadPreference.add_read_preference(cmd, opts)
-
     ## check, if retryable reads are enabled
     opts = Mongo.retryable_reads(opts)
 
     with {:ok, session} <- Session.start_implicit_session(topology_pid, :read, opts),
-         result <- exec_command_session(session, new_cmd, opts),
+         result <- exec_command_session(session, cmd, opts),
          :ok <- Session.end_implict_session(topology_pid, session) do
       case result do
         {:error, error} ->

lib/mongo/read_preference.ex

@@ -37,17 +37,17 @@ defmodule Mongo.ReadPreference do
     hedge: BSON.document
   }
 
-  @default %{
+  @primary %{
     mode: :primary,
     tag_sets: [],
     max_staleness_ms: 0
   }
 
-  def defaults(map \\ nil)
-  def defaults(map) when is_map(map) do
-    Map.merge(@default, map)
+  def primary(map \\ nil)
+  def primary(map) when is_map(map) do
+    Map.merge(@primary, map)
   end
-  def defaults(_), do: @default
+  def primary(_), do: @primary
 
   @doc """
   Add read preference to the cmd
@@ -56,7 +56,7 @@ defmodule Mongo.ReadPreference do
 
     read_preference = opts
                       |> Keyword.get(:read_preference)
-                      |> Mongo.ReadPreference.defaults()
+                      |> Mongo.ReadPreference.primary()
                       |> transform()
 
     cmd ++ ["$readPreference": read_preference]
@@ -86,4 +86,32 @@ defmodule Mongo.ReadPreference do
 
   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
+    nil
+  end
+  # For mode 'secondary', drivers MUST set the slaveOK wire protocol flag and MUST also use $readPreference
+  def mongos(%{mode: :secondary} = config) do
+    transform(config)
+  end
+  # For mode 'primaryPreferred', drivers MUST set the slaveOK wire protocol flag and MUST also use $readPreference
+  def mongos(%{mode: :primary_preferred} = config) do
+    transform(config)
+  end
+  # For mode 'secondaryPreferred', drivers MUST set the slaveOK wire protocol flag. If the read preference contains a
+  # non-empty tag_sets parameter, maxStalenessSeconds is a positive integer, or the hedge parameter is non-empty,
+  # drivers MUST use $readPreference; otherwise, drivers MUST NOT use $readPreference
+  def mongos(%{mode: :secondary_preferred} = config) do
+    transform(config)
+  end
+  # For mode 'nearest', drivers MUST set the slaveOK wire protocol flag and MUST also use $readPreference
+  def mongos(%{mode: :nearest} = config) do
+    transform(config)
+  end
+
+
+
 end

lib/mongo/topology.ex

@@ -253,21 +253,22 @@ defmodule Mongo.Topology do
     {:reply, Map.fetch(state.connection_pools, address), state}
   end
 
+  ##
   # checkout a new session
+  #
   def handle_call({:checkout_session, cmd_type, type, opts}, from, %{:topology => topology, :waiting_pids => waiting} = state) do
 
     case TopologyDescription.select_servers(topology, cmd_type, opts) do
       :empty ->
         Mongo.Events.notify(%ServerSelectionEmptyEvent{action: :checkout_session, cmd_type: cmd_type, topology: topology, opts: opts})
         {:noreply, %{state | waiting_pids: [from | waiting]}} ## no servers available, wait for connection
 
-      {:ok, servers} ->                ## found, select randomly a server and return its connection_pool
-        with address <- Enum.take_random(servers, 1),
-             {:ok, connection} <- get_connection(address, state),
-              wire_version <- wire_version(address, topology),
-            {server_session, new_state} <- checkout_server_session(state),
-            {:ok, session} <- Session.start_link(self(), connection, server_session, type, wire_version, opts) do
-
+      ## found
+      {:ok, {address, opts}} ->
+        with {:ok, connection}           <- get_connection(address, state),
+             wire_version                <- wire_version(address, topology),
+             {server_session, new_state} <- checkout_server_session(state),
+             {:ok, session}              <- Session.start_link(self(), connection, server_session, type, wire_version, opts) do
           {:reply, {:ok, session}, new_state}
         end
 
@@ -283,10 +284,8 @@ defmodule Mongo.Topology do
         Mongo.Events.notify(%ServerSelectionEmptyEvent{action: :select_server, cmd_type: cmd_type, topology: topology, opts: opts})
         {:noreply, %{state | waiting_pids: [from | waiting]}} ## no servers available, wait for connection
 
-      {:ok, servers} ->                ## found, select randomly a server and return its connection_pool
-        with {:ok, connection} <- servers
-                                  |> Enum.take_random(1)
-                                  |> get_connection(state) do
+      {:ok, {address, _opts}} ->
+        with {:ok, connection} <- get_connection(address, state) do
           {:reply, {:ok, connection}, state}
         end
       error ->
@@ -299,10 +298,8 @@ defmodule Mongo.Topology do
       :empty ->
         Mongo.Events.notify(%ServerSelectionEmptyEvent{action: :limits, cmd_type: :write, topology: topology})
         {:reply, nil, state}
-      {:ok, servers} ->                ## found, select randomly a server and return its connection_pool
-        with {:ok, limits} <- servers
-                              |> Enum.take_random(1)
-                              |> get_limits(topology) do
+      {:ok, {address, _opts}} ->
+        with {:ok, limits} <- get_limits(address, topology) do
           {:reply, {:ok, limits}, state}
         end
       error ->
@@ -316,10 +313,8 @@ defmodule Mongo.Topology do
         Mongo.Events.notify(%ServerSelectionEmptyEvent{action: :wire_version, cmd_type: :write, topology: topology})
         {:reply, nil, state}
 
-      {:ok, servers} ->                ## found, select randomly a server and return its connection_pool
-        with address <- Enum.take_random(servers, 1) do
-          {:reply, {:ok, wire_version(address, topology)}, state}
-        end
+      {:ok, {address, _opts}} ->
+        {:reply, {:ok, wire_version(address, topology)}, state}
       error ->
         {:reply, error, state} ## in case of an error, just return the error
     end

lib/mongo/topology_description.ex

@@ -69,28 +69,49 @@ defmodule Mongo.TopologyDescription do
   def select_servers(%{:compatible => false}, _type, _opts) do
     {:error, :invalid_wire_version}
   end
-  def select_servers(topology, type, opts) do
+  def select_servers(topology, :write, opts) do
+    servers = case topology.type do
+      :single                   -> topology.servers
+      :sharded                  -> mongos_servers(topology)
+      :replica_set_with_primary -> primary_servers(topology)
+      _                         -> []
+    end
+
+    addr = servers
+           |> Enum.map(fn {server, _} -> server end)
+           |> Enum.take_random(1)
+
+    case addr do
+      [] -> :empty
+      _  -> {:ok, {addr, opts}}
+    end
+  end
+  def select_servers(topology, :read, opts) do
 
     read_preference = opts
                       |> Keyword.get(:read_preference)
-                      |> ReadPreference.defaults()
+                      |> ReadPreference.primary()
 
-    servers = case topology.type do
-      :unknown -> []
-      :single  -> topology.servers
-      :sharded -> mongos_servers(topology)
-      _        ->
-        case {type, topology.type == :replica_set_with_primary} do
-          {:read, _}     -> select_replica_set_server(topology, read_preference.mode, read_preference)
-          {:write, true} -> select_replica_set_server(topology, :primary, ReadPreference.defaults)
-          _              ->  []
-        end
+    {servers, read_prefs} = case topology.type 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}
     end
 
+    opts = case read_prefs do
+      nil   -> Keyword.delete(opts, :read_preference)
+      prefs -> Keyword.put(opts, :read_preference, prefs)
+    end
+
+    addr = servers
+           |> Enum.map(fn {server, _} -> server end)
+           |> Enum.take_random(1)
+
     # check now three possible cases
-    case Enum.map(servers, fn {server, _} -> server end) do
-      []        -> :empty
-      servers   -> {:ok, servers}
+    case addr do
+      [] -> :empty
+      _  -> {:ok, {addr, opts}}
     end
   end
 

test/mongo/topology_description_test.exs

@@ -7,66 +7,75 @@ defmodule Mongo.TopologyDescriptionTest do
     single_server = ["localhost:27017"]
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :secondary})
+      read_preference: ReadPreference.primary(%{mode: :secondary})
     ]
-    assert {:ok, single_server} == TopologyDescription.select_servers(single(), :read, opts)
+    assert {:ok, {^single_server, _}} = TopologyDescription.select_servers(single(), :read, opts)
 
-    assert {:ok, single_server} ==TopologyDescription.select_servers(single(), :write)
+    assert {:ok, {^single_server, _}} = TopologyDescription.select_servers(single(), :write)
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :nearest})
+      read_preference: ReadPreference.primary(%{mode: :nearest})
     ]
-    assert {:ok, single_server} ==TopologyDescription.select_servers(single(), :read, opts)
+    assert {:ok, {^single_server, _}} = TopologyDescription.select_servers(single(), :read, opts)
   end
 
   test "replica set server selection" do
     all_hosts = ["localhost:27018", "localhost:27019", "localhost:27020"]
     master = "localhost:27018"
+    seconardaries = List.delete(all_hosts, master)
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :secondary})
+      read_preference: ReadPreference.primary(%{mode: :secondary})
     ]
-    assert {:ok, List.delete(all_hosts, master)} == TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
+    {:ok, {[server], _}} = TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
+
+    assert Enum.any?(seconardaries, fn sec -> sec == server end)
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :primary})
+      read_preference: ReadPreference.primary(%{mode: :primary})
     ]
-    assert {:ok, [master]} == TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
+    assert {:ok, {[master], _}} = TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :primary_preferred})
+      read_preference: ReadPreference.primary(%{mode: :primary_preferred})
     ]
-    assert {:ok, [master]} == TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
+    assert {:ok, {[master], _}} = TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :primary_preferred})
+      read_preference: ReadPreference.primary(%{mode: :primary_preferred})
     ]
-    assert {:ok, List.delete(all_hosts, master)} == TopologyDescription.select_servers(repl_set_no_master(), :read, opts)
+    {:ok, {[server], _}} = TopologyDescription.select_servers(repl_set_no_master(), :read, opts)
+    assert Enum.any?(seconardaries, fn sec -> sec == server end)
 
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :nearest})
+      read_preference: ReadPreference.primary(%{mode: :nearest})
     ]
-    assert {:ok, all_hosts} == TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
+    {:ok, {[server], _}} = TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
+    assert Enum.any?(all_hosts, fn sec -> sec == server end)
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :secondary})
+      read_preference: ReadPreference.primary(%{mode: :secondary})
     ]
-    assert {:ok, List.delete(all_hosts, master)} == TopologyDescription.select_servers(repl_set_no_master(), :read, opts)
+    {:ok, {[server], _}} = TopologyDescription.select_servers(repl_set_no_master(), :read, opts)
+    assert Enum.any?(seconardaries, fn sec -> sec == server end)
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :secondary_preferred})
+      read_preference: ReadPreference.primary(%{mode: :secondary_preferred})
     ]
-    assert {:ok, List.delete(all_hosts, master)} == TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
+    {:ok, {[server], _}} = TopologyDescription.select_servers(repl_set_with_master(), :read, opts)
+    assert Enum.any?(seconardaries, fn sec -> sec == server end)
 
-    assert {:ok, [master]} == TopologyDescription.select_servers(repl_set_only_master(), :read, opts)
+    assert {:ok, {[^master], _}} = TopologyDescription.select_servers(repl_set_only_master(), :read, opts)
 
-    assert {:ok, List.delete(all_hosts, master)} == TopologyDescription.select_servers(repl_set_no_master(), :read, opts)
+    {:ok, {[server], _}} = TopologyDescription.select_servers(repl_set_no_master(), :read, opts)
+    assert Enum.any?(seconardaries, fn sec -> sec == server end)
 
     opts = [
-      read_preference: ReadPreference.defaults(%{mode: :nearest})
+      read_preference: ReadPreference.primary(%{mode: :nearest})
     ]
-    assert {:ok, all_hosts} == TopologyDescription.select_servers(repl_set_no_master(), :read, opts)
+    {:ok, {[server], _}} = TopologyDescription.select_servers(repl_set_no_master(), :read, opts)
+    assert Enum.any?(all_hosts, fn sec -> sec == server end)
 
   end
 
@@ -76,6 +85,6 @@ defmodule Mongo.TopologyDescriptionTest do
     opts = [
       read_preference: %{mode: :secondary}
     ]
-    assert {:ok, single_server} == TopologyDescription.select_servers(single(), :read, opts)
+    assert {:ok, {^single_server, _}} = TopologyDescription.select_servers(single(), :read, opts)
   end
 end

test/mongo/topology_test.exs

@@ -11,7 +11,7 @@ defmodule Mongo.TopologyTest do
     for mode <- @modes do
       assert {:ok, %Mongo.InsertOneResult{inserted_id: new_id}} = Mongo.insert_one(mongo_pid, "test", %{topology_test: 1}, w: 3)
 
-      rp = Mongo.ReadPreference.defaults(%{mode: mode})
+      rp = Mongo.ReadPreference.primary(%{mode: mode})
       assert [%{"_id" => ^new_id, "topology_test" => 1}] =
                mongo_pid
                |> Mongo.find("test", %{_id: new_id}, read_preference: rp, slave_ok: mode in [:secondary, :secondary_preferred])