fixes #31

Author: zookzook

lib/mongo.ex

@@ -630,23 +630,40 @@ defmodule Mongo do
   defp check_for_error(%{"code" => code, "errmsg" => msg}), do: {:error, Mongo.Error.exception(message: msg, code: code)}
 
   @doc """
-  Returns the current wire version.
+  Returns the wire version of the database
+  ## Example
+
+      {:ok, top} = Mongo.start_link(...)
+      Mongo.wire_version(top)
+
+      {:ok, 8}
   """
   @spec wire_version(pid) :: {:ok, integer} | {:error, Mongo.Error.t}
-  def wire_version(conn) do
-    cmd = %Query{action: :wire_version}
-    with {:ok, _cmd, version} <- DBConnection.execute(conn, cmd, %{}, defaults([])) do
-      {:ok, version}
+  def wire_version(topology_pid) do
+    with {:ok, wire_version} <- Topology.wire_version(topology_pid) do
+      {:ok, wire_version}
     end
   end
 
   @doc """
   Returns the limits of the database.
+
+  ## Example
+
+      {:ok, top} = Mongo.start_link(...)
+      Mongo.limits(top)
+
+      {:ok, %{
+         logical_session_timeout: 30,
+         max_bson_object_size: 16777216,
+         max_message_size_bytes: 48000000,
+         max_wire_version: 8,
+         max_write_batch_size: 100000
+      }}
   """
   @spec limits(pid) :: {:ok, BSON.document} | {:error, Mongo.Error.t}
-  def limits(conn) do
-    cmd = %Query{action: :limits}
-    with {:ok, _cmd, limits} <- DBConnection.execute(conn, cmd, %{}, defaults([])) do
+  def limits(topology_pid) do
+    with {:ok, limits} <- Topology.limits(topology_pid) do
       {:ok, limits}
     end
   end

lib/mongo/bulk_write.ex

@@ -183,7 +183,7 @@ defmodule Mongo.BulkWrite do
   def write(topology_pid, %UnorderedBulk{} = bulk, opts) do
 
     with {:ok, session} <- Session.start_implicit_session(topology_pid, :write, opts),
-         result         = one_bulk_write(session, bulk, opts),
+         result         = one_bulk_write(topology_pid, session, bulk, opts),
          :ok            <- Session.end_implict_session(topology_pid, session) do
       result
       else
@@ -201,7 +201,7 @@ defmodule Mongo.BulkWrite do
     empty = %BulkWriteResult{acknowledged: acknowledged?(write_concern)}
 
     with {:ok, session} <- Session.start_implicit_session(topology_pid, :write, opts),
-         {:ok, limits} <- get_limits(session),
+         {:ok, limits} <- Mongo.limits(topology_pid),
          max_batch_size <- limits.max_write_batch_size,
          result = ops
                   |> get_op_sequence()
@@ -227,9 +227,9 @@ defmodule Mongo.BulkWrite do
   # The function returns a keyword list with the results of each operation group:
   # For the details see https://github.com/mongodb/specifications/blob/master/source/crud/crud.rst#results
   #
-  defp one_bulk_write(session, %UnorderedBulk{coll: coll, inserts: inserts, updates: updates, deletes: deletes}, opts) do
+  defp one_bulk_write(topology_pid, session, %UnorderedBulk{coll: coll, inserts: inserts, updates: updates, deletes: deletes}, opts) do
 
-    with {:ok, limits} <- get_limits(session),
+    with {:ok, limits} <- Mongo.limits(topology_pid),
          max_batch_size <- limits.max_write_batch_size,
          insert_result <- one_bulk_write_operation(session, :insert, coll, inserts, max_batch_size, opts),
          update_result <- one_bulk_write_operation(session, :update, coll, updates, max_batch_size, opts),
@@ -418,10 +418,4 @@ defmodule Mongo.BulkWrite do
 
   end
 
-  defp get_limits(session) do
-    with conn <- Session.connection(session) do
-      Mongo.limits(conn)
-    end
-  end
-
 end

lib/mongo/cursor.ex

@@ -104,7 +104,7 @@ defmodule Mongo.Cursor do
                "id" => cursor_id,
                "ns" => coll,
                "firstBatch" => docs} = response}} when ok == 1 <- Mongo.exec_command_session(session, new_cmd, opts),
-           {:ok, wire_version} <- wire_version(session) do
+           {:ok, wire_version} <- Mongo.wire_version(topology_pid) do
 
           [%{"$changeStream" => stream_opts} | _pipeline] = Keyword.get(new_cmd, :pipeline) # extract the change stream options
 
@@ -183,7 +183,7 @@ defmodule Mongo.Cursor do
         {:error, %Mongo.Error{code: code} = not_resumable} when code == 11601 or code == 136 or code == 237 -> {:error, not_resumable}
         {:error, _error} ->
 
-          with {:ok, wire_version} <- wire_version(session) do
+          with {:ok, wire_version} <- Mongo.wire_version(topology_pid) do
 
             [%{"$changeStream" => stream_opts} | pipeline] = Keyword.get(aggregate_cmd, :pipeline) # extract the change stream options
 
@@ -319,12 +319,6 @@ defmodule Mongo.Cursor do
     def count(_stream), do: {:error, __MODULE__}
     def member?(_stream, _term), do: {:error, __MODULE__}
 
-    defp wire_version(session) do
 
-      session
-      |> Mongo.Session.connection()
-      |> Mongo.wire_version()
-
-    end
   end
 end

lib/mongo/mongo_db_connection.ex

@@ -28,8 +28,7 @@ defmodule Mongo.MongoDBConnection do
       connect_timeout: opts[:connect_timeout] || @timeout,
       database: Keyword.fetch!(opts, :database),
       write_concern: Map.new(write_concern),
-      wire_version: nil, # todo move to topolgy and topology-description
-      limits: nil, # todo  move to topolgy and topology-description
+      wire_version: 0,
       auth_mechanism: opts[:auth_mechanism] || nil,
       connection_type: Keyword.fetch!(opts, :connection_type),
       topology_pid: Keyword.fetch!(opts, :topology_pid),
@@ -128,7 +127,7 @@ defmodule Mongo.MongoDBConnection do
     cmd = [ismaster: 1, client: client] |> filter_nils()
 
     case Utils.command(-1, cmd, state) do
-      {:ok, %{"ok" => ok, "maxWireVersion" => version} = response}  when ok == 1 -> {:ok, %{update_limits(state, response) | wire_version: version}}
+      {:ok, %{"ok" => ok, "maxWireVersion" => version}}  when ok == 1 -> {:ok, %{state | wire_version: version}}
       {:ok, %{"ok" => ok}} when ok == 1 ->  {:ok, %{state | wire_version: 0}}
       {:ok, %{"ok" => ok, "errmsg" => msg, "code" => code}} when ok == 0 ->
         err = Mongo.Error.exception(message: msg, code: code)
@@ -188,29 +187,6 @@ defmodule Mongo.MongoDBConnection do
   defp pretty_name("apple"), do: "Mac OS X"
   defp pretty_name(name), do: name
 
-  ##
-  #
-  # Updates the limits from the isMaster response:
-  #
-  # isMaster.maxBsonObjectSize / 16 * 1024 * 1024
-  # isMaster.maxMessageSizeBytes / 48000000
-  # isMaster.maxWriteBatchSize /  100,000
-  # isMaster.localTime
-  # isMaster.compression
-  # isMaster.readOnly
-  # isMaster.logicalSessionTimeoutMinutes
-  #
-  defp update_limits(state, result) do
-    limits = %{max_bson_object_size: (result["maxBsonObjectSize"] || 16_777_216),
-      max_message_size_bytes: (result["maxMessageSizeBytes"] || 48_000_000),
-      max_write_batch_size: (result["maxWriteBatchSize"] || 100_000),
-      compression: result["compression"],
-      read_only: (result["readOnly"] || false),
-      logical_session_timeout_minutes: result["logicalSessionTimeoutMinutes"]}
-
-    %{state | limits: limits}
-  end
-
   @impl true
   def checkout(state), do: {:ok, state}
   @impl true
@@ -252,12 +228,6 @@ defmodule Mongo.MongoDBConnection do
     end
   end
 
-  defp execute_action(:limits, _, _, state) do
-    {:ok, state.limits, state}
-  end
-  defp execute_action(:wire_version, _, _, state) do
-    {:ok, state.wire_version, state}
-  end
   defp execute_action(:command, [cmd], opts, %{wire_version: version} = state) when version >= 6 do
 
     cmd = cmd ++ ["$db": opts[:database] || state.database]

lib/mongo/server_description.ex

@@ -23,6 +23,11 @@ defmodule Mongo.ServerDescription do
     election_id: BSON.ObjectId.t | nil,
     primary: String.t | nil,
     last_update_time: non_neg_integer,
+    max_bson_object_size: non_neg_integer,
+    max_message_size_bytes: non_neg_integer,
+    max_write_batch_size: non_neg_integer,
+    compression: String.t | nil,
+    read_only: boolean(),
     logical_session_timeout: non_neg_integer
   }
 
@@ -46,6 +51,11 @@ defmodule Mongo.ServerDescription do
       election_id: nil,
       primary: nil,
       last_update_time: 0,
+      max_bson_object_size: 16_777_216,
+      max_message_size_bytes: 48_000_000,
+      max_write_batch_size: 100_000,
+      compression: nil,
+      read_only: false,
       logical_session_timeout: 30
     }, map)
   end
@@ -77,6 +87,11 @@ defmodule Mongo.ServerDescription do
       set_version: is_master_reply["setVersion"],
       election_id: is_master_reply["electionId"],
       primary: is_master_reply["primary"],
+      max_bson_object_size: (is_master_reply["maxBsonObjectSize"] || 16_777_216),
+      max_message_size_bytes: (is_master_reply["maxMessageSizeBytes"] || 48_000_000),
+      max_write_batch_size: (is_master_reply["maxWriteBatchSize"] || 100_000),
+      compression: is_master_reply["compression"],
+      read_only: (is_master_reply["readOnly"] || false),
       logical_session_timeout: is_master_reply["logicalSessionTimeoutMinutes"] || 30
     }
   end

lib/mongo/topology.ex

@@ -12,6 +12,8 @@ defmodule Mongo.Topology do
   alias Mongo.Session.SessionPool
   alias Mongo.Session
 
+  @limits [:logical_session_timeout, :max_bson_object_size, :max_message_size_bytes, :max_wire_version, :max_write_batch_size]
+
   # https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#heartbeatfrequencyms-defaults-to-10-seconds-or-60-seconds
   @heartbeat_frequency_ms 10_000
 
@@ -52,6 +54,14 @@ defmodule Mongo.Topology do
     GenServer.call(pid, {:select_server, type, opts})
   end
 
+  def limits(pid) do
+    GenServer.call(pid, :limits)
+  end
+
+  def wire_version(pid) do
+    GenServer.call(pid, :wire_version)
+  end
+
   @doc """
 
   """
@@ -248,9 +258,9 @@ defmodule Mongo.Topology do
       {:ok, servers} ->                ## found, select randomly a server and return its connection_pool
         Logger.debug("select_server: found #{inspect servers}, pools: #{inspect pools}")
 
-        with host <- Enum.take_random(servers, 1),
-             {:ok, connection} <- get_connection(host, state),
-              wire_version <- max_version(topology, host),
+        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(connection, server_session, type, wire_version, opts) do
 
@@ -264,14 +274,14 @@ defmodule Mongo.Topology do
     end
   end
 
-  def handle_call({:select_server, type, opts}, from, %{:topology => topology, :waiting_pids => waiting, connection_pools: pools} = state) do
+  def handle_call({:select_server, type, opts}, from, %{:topology => topology, :waiting_pids => waiting} = state) do
     case TopologyDescription.select_servers(topology, type, opts) do
       :empty ->
         Logger.debug("select_server: empty")
         {:noreply, %{state | waiting_pids: [from | waiting]}} ## no servers available, wait for connection
 
       {:ok, servers} ->                ## found, select randomly a server and return its connection_pool
-        Logger.debug("select_server: found #{inspect servers}, pools: #{inspect pools}")
+        Logger.info("select_server: found #{inspect servers}")
 
         with {:ok, connection} <- servers
                                   |> Enum.take_random(1)
@@ -286,6 +296,45 @@ defmodule Mongo.Topology do
     end
   end
 
+  def handle_call(:limits, _from, %{:topology => topology} = state) do
+    case TopologyDescription.select_servers(topology, :write, []) do
+      :empty ->
+        Logger.debug("select_server: empty")
+        {:reply, nil, state}
+
+      {:ok, servers} ->                ## found, select randomly a server and return its connection_pool
+        Logger.debug("select_server: found #{inspect servers}")
+
+        with {:ok, limits} <- servers
+                              |> Enum.take_random(1)
+                              |> get_limits(topology) do
+          Logger.debug("select_server: connection is #{inspect limits}")
+
+          {:reply, {:ok, limits}, state}
+        end
+      error ->
+        Logger.debug("select_servers: #{inspect error}")
+        {:reply, error, state} ## in case of an error, just return the error
+    end
+  end
+
+  def handle_call(:wire_version, _from, %{:topology => topology} = state) do
+    case TopologyDescription.select_servers(topology, :write, []) do
+      :empty ->
+        Logger.debug("select_server: empty")
+        {:reply, nil, state}
+
+      {:ok, servers} ->                ## found, select randomly a server and return its connection_pool
+        Logger.debug("select_server: found #{inspect servers}")
+        with address <- Enum.take_random(servers, 1) do
+          {:reply, {:ok, wire_version(address, topology)}, state}
+        end
+      error ->
+        Logger.debug("select_servers: #{inspect error}")
+        {:reply, error, state} ## in case of an error, just return the error
+    end
+  end
+
   defp checkout_server_session(%{:session_pool => session_pool} = state) do
     with {session, pool} <- SessionPool.checkout(session_pool) do
       {session, %{state | session_pool: pool}}
@@ -298,10 +347,19 @@ defmodule Mongo.Topology do
   defp get_connection([], _state), do: nil
   defp get_connection([address], %{connection_pools: pools}), do: Map.fetch(pools, address)
 
-  defp max_version(topology, [host]) do
-    case Map.get(topology.servers, host) do
-      nil    -> 0
-      server -> server.max_wire_version
+  defp get_limits([], _topology), do: nil
+  defp get_limits([address], %{servers: servers}) do
+    with {:ok, desc} <- Map.fetch(servers, address) do
+      {:ok, Map.take(desc, @limits)}
+    end
+  end
+
+  defp wire_version([], _topology), do: nil
+  defp wire_version([address], topology) do
+    with {:ok, server} <- Map.fetch(topology.servers, address) do
+      server.max_wire_version
+    else
+      _other -> 0
     end
   end