fixes #54

Author: zookzook

README.md

@@ -52,13 +52,9 @@ BSON symbols can only be decoded.
 
 ### Installation:
 
-Add `mongodb_driver` to your mix.exs `deps` and `:applications`.
+Add `mongodb_driver` to your mix.exs `deps`.
 
 ```elixir
-def application do
-  [applications: [:mongodb_driver]]
-end
-
 defp deps do
   [{:mongodb_driver, "~> 0.6"}]
 end
@@ -171,6 +167,27 @@ you'll want to add this cipher to your `ssl_opts`:
       ]
 )
 ```
+
+### Find
+
+Using `$and`
+
+```elixir
+Mongo.find(:mongo, "users", %{"$and" => [%{email: "[email protected]"}, %{first_name: "first_name"}]})
+```
+
+Using `$or`
+
+```elixir
+Mongo.find(:mongo, "users", %{"$or" => [%{email: "[email protected]"}, %{first_name: "first_name"}]})
+```
+
+Using `$in`
+
+```elixir
+Mongo.find(:mongo, "users", %{email: %{"$in" => ["[email protected]", "[email protected]"]}})
+```
+
 ### Change streams
 
 Change streams exist in replica set and cluster systems and tell you about changes to collections. 
@@ -351,25 +368,10 @@ For more information see:
 
 and have a look at the test units as well.
 
-### Examples
 
-Using `$and`
-
-```elixir
-Mongo.find(:mongo, "users", %{"$and" => [%{email: "[email protected]"}, %{first_name: "first_name"}]})
-```
+### Command Monitoring
 
-Using `$or`
-
-```elixir
-Mongo.find(:mongo, "users", %{"$or" => [%{email: "[email protected]"}, %{first_name: "first_name"}]})
-```
-
-Using `$in`
-
-```elixir
-Mongo.find(:mongo, "users", %{email: %{"$in" => ["[email protected]", "[email protected]"]}})
-```
+You can watch all events that are triggered while the driver send requests and processes responses.
 
 ## Testing and Travis CI
 
@@ -410,32 +412,17 @@ $ mongod --sslMode allowSSL --sslPEMKeyFile /path/to/mongodb.pem
 
 Special thanks to [JetBrains](https://www.jetbrains.com/?from=elixir-mongodb-driver) for providing a free JetBrains Open Source license for their complete toolbox.
 
-This is an alternative development from the [original](https://github.com/ankhers/mongodb), which was the starting point
-and already contained very nice code.
-
 The [Documentation](https://hexdocs.pm/mongodb_driver/readme.html) is online, but currently not up to date. 
 This will be done as soon as possible. In the meantime, look in the source code. Especially 
 for the individual options.  
 
-## Motivation
-
-  * [x] I have made a number of changes to understand how the driver works. For example, I reduced cursor modules to just one cursor and
-        replaced some op code calls with command calls.
-  * [x] Simplify code: remove raw_find (raw_find called from cursors, raw_find called with "$cmd"), so raw_find is more calling a command than a find query.
-  * [x] Better support for new MongoDB version, for example the ability to use views
-  * [x] Upgraded to ([DBConnection 2.x](https://github.com/elixir-ecto/db_connection))
-  * [x] Removed depreacated op codes ([See](https://docs.mongodb.com/manual/reference/mongodb-wire-protocol/#request-opcodes))
-  * [x] Added `op_msg` support ([See](https://docs.mongodb.com/manual/reference/mongodb-wire-protocol/#op-msg))
-  * [x] Added bulk writes ([See](https://github.com/mongodb/specifications/blob/master/source/crud/crud.rst#write))
-  * [x] Add support for driver sessions ([See](https://github.com/mongodb/specifications/blob/master/source/sessions/driver-sessions.rst))
-  * [x] Add support for driver transactions ([See](https://github.com/mongodb/specifications/blob/master/source/transactions/transactions.rst))
-  * [ ] Add support for `op_compressed` ([See](https://github.com/mongodb/specifications/blob/master/source/compression/OP_COMPRESSED.rst))
-  
+This driver is based on [original](https://github.com/ankhers/mongodb).
+ 
 ## License
 
 Copyright 2015 Eric Meadows-Jönsson and Justin Wood
 
-Copyright 2019 Michael Maier
+Copyright 2019 - 2020 Michael Maier
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

lib/mongo/event_handler.ex

@@ -3,7 +3,7 @@ defmodule Mongo.EventHandler do
 
   require Logger
 
-  @all [:commands, :is_master, :topology]
+  @all [:commands, :topology]
 
   def start(opts \\ [:commands]) do
     Logger.info("Starting EventHandler")

lib/mongo/events.ex

@@ -83,6 +83,15 @@ defmodule Mongo.Events do
     ]
   end
 
+
+  ##
+  #
+  defmodule ServerSelectionEmptyEvent do
+    @moduledoc false
+    defstruct [:action, :cmd_type, :topology, :opts]
+  end
+
+
   # Published when server description changes, but does NOT include changes to
   # the RTT
   defmodule ServerDescriptionChangedEvent do

lib/mongo/monitor.ex

@@ -164,13 +164,16 @@ defmodule Mongo.Monitor do
 
   defp call_is_master_command(conn_pid, opts) do
 
-    start_time  = System.monotonic_time
-    result      = Mongo.exec_command(conn_pid, [isMaster: 1], opts)
-    finish_time = System.monotonic_time
-    rtt         = System.convert_time_unit(finish_time - start_time, :native, :millisecond)
-    finish_time = System.convert_time_unit(finish_time, :native, :millisecond)
+    {rtt, result} = :timer.tc(fn -> Mongo.exec_command(conn_pid, [isMaster: 1], opts) end)
 
-    {result, finish_time, rtt}
+    finish_time = DateTime.utc_now()
+    #start_time  = System.monotonic_time
+    #result      = Mongo.exec_command(conn_pid, [isMaster: 1], opts)
+    #finish_time = System.monotonic_time
+    #rtt         = System.convert_time_unit(finish_time - start_time, :native, :millisecond)
+    #finish_time = System.convert_time_unit(finish_time, :native, :millisecond)
+
+    {result, finish_time, div(rtt, 1000)}
   end
 
   defp notify_error(rtt, error, conn_pid) do

lib/mongo/read_preference.ex

@@ -1,17 +1,45 @@
 defmodule Mongo.ReadPreference do
+
+  import Keywords
+
   @moduledoc ~S"""
   Determines which servers are considered suitable for read operations
+
+  A read preference consists of a mode and optional `tag_sets`, max_staleness_ms, and `hedge`.
+  The mode prioritizes between primaries and secondaries to produce either a single suitable server or a list of candidate servers.
+  If tag_sets and maxStalenessSeconds are set, they determine which candidate servers are eligible for selection.
+  If hedge is set, it configures how server hedged reads are used.
+
+  The default mode is 'primary'.
+  The default tag_sets is a list with an empty tag set: [{}].
+  The default max_staleness_ms is unset.
+  The default hedge is unset.
+
+  ## mode
+
+  * `primary` Only an available primary is suitable.
+  * `secondary` All secondaries (and only secondaries) are candidates, but only eligible candidates (i.e. after applying tag_sets and maxStalenessSeconds) are suitable.
+  * `primaryPreferred` If a primary is available, only the primary is suitable. Otherwise, all secondaries are candidates,
+                       but only eligible secondaries are suitable.
+  * `secondaryPreferred` All secondaries are candidates. If there is at least one eligible secondary, only eligible secondaries are suitable.
+                         Otherwise, when there are no eligible secondaries, the primary is suitable.
+  * `nearest` The primary and all secondaries are candidates, but only eligible candidates are suitable.
+
   """
   @type t :: %{
-    mode: :primary | :secondary | :primary_preferred | :secondary_preferred |
+    mode: :primary |
+          :secondary |
+          :primary_preferred |
+          :secondary_preferred |
           :nearest,
     tag_sets: [%{String.t => String.t}],
-    max_staleness_ms: non_neg_integer
+    max_staleness_ms: non_neg_integer,
+    hedge: BSON.document
   }
 
   @default %{
     mode: :primary,
-    tag_sets: [%{}],
+    tag_sets: [],
     max_staleness_ms: 0
   }
 
@@ -29,13 +57,38 @@ defmodule Mongo.ReadPreference do
     read_preference = opts
                       |> Keyword.get(:read_preference)
                       |> Mongo.ReadPreference.defaults()
-                      |> Map.update(:mode, :primary, fn mode -> map_mode(mode) end)
+                      |> transform()
 
     cmd ++ ["$readPreference": read_preference]
   end
 
-  defp map_mode(:primary_preferred), do: :primaryPreferred
-  defp map_mode(:secondary_preferred), do: :secondaryPreferred
-  defp map_mode(mode), do: mode
+  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
+
+  defp is_max_staleness_valid?() do
+    #max_staleness_ms >= heartbeatFrequencyMS + idleWritePeriodMS
+    #max_staleness_ms >= smallestMaxStalenessSeconds
+
+  end
 end

lib/mongo/topology.ex

@@ -5,7 +5,8 @@ defmodule Mongo.Topology do
 
   use GenServer
   alias Mongo.Events.{ServerDescriptionChangedEvent, ServerOpeningEvent, ServerClosedEvent,
-                      TopologyDescriptionChangedEvent, TopologyOpeningEvent, TopologyClosedEvent}
+                      TopologyDescriptionChangedEvent, TopologyOpeningEvent, TopologyClosedEvent,
+                      ServerSelectionEmptyEvent}
   alias Mongo.TopologyDescription
   alias Mongo.ServerDescription
   alias Mongo.Monitor
@@ -99,7 +100,8 @@ defmodule Mongo.Topology do
               type: type,
               set_name: set_name,
               servers: servers,
-              local_threshold_ms: local_threshold_ms
+              local_threshold_ms: local_threshold_ms,
+              heartbeat_frequency_ms: @heartbeat_frequency_ms
             }),
             seeds: seeds,
             opts: opts,
@@ -252,23 +254,20 @@ defmodule Mongo.Topology do
   end
 
   # checkout a new session
-  def handle_call({:checkout_session, cmd_type, type, opts}, from, %{:topology => topology, :waiting_pids => waiting, connection_pools: pools} = state) do
+  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 ->
-        Logger.debug("select_server: 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
-        Logger.debug("select_server: found #{inspect servers}, pools: #{inspect pools}")
-
         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
 
-          Logger.debug("select_server: connection is #{inspect connection}, server_session is #{inspect server_session}")
           {:reply, {:ok, session}, new_state}
         end
 
@@ -278,63 +277,50 @@ defmodule Mongo.Topology do
     end
   end
 
-  def handle_call({:select_server, type, opts}, from, %{:topology => topology, :waiting_pids => waiting} = state) do
-    case TopologyDescription.select_servers(topology, type, opts) do
+  def handle_call({:select_server, cmd_type, opts}, from, %{:topology => topology, :waiting_pids => waiting} = state) do
+    case TopologyDescription.select_servers(topology, cmd_type, opts) do
       :empty ->
-        Logger.debug("select_server: empty")
+        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
-        Logger.debug("select_server: found #{inspect servers}")
-
         with {:ok, connection} <- servers
                                   |> Enum.take_random(1)
                                   |> get_connection(state) do
-          Logger.debug("select_server: connection is #{inspect connection}")
-
           {:reply, {:ok, connection}, 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(:limits, _from, %{:topology => topology} = state) do
     case TopologyDescription.select_servers(topology, :write, []) do
       :empty ->
-        Logger.debug("select_server: 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
-        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")
+        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
-        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