#63: fixed wire version in change streams, removed some compiler warnings

Author: zookzook

lib/mongo/bulk_write.ex

@@ -191,7 +191,7 @@ defmodule Mongo.BulkWrite do
 
   end
 
-  def write(topology_pid, %OrderedBulk{coll: coll, ops: ops} = bulk, opts) do
+  def write(topology_pid, %OrderedBulk{coll: coll, ops: ops}, opts) do
 
     write_concern = write_concern(opts)
 

lib/mongo/change_stream.ex

@@ -3,8 +3,6 @@ defmodule Mongo.ChangeStream do
   alias Mongo.Session
   alias Mongo.Error
 
-  require Logger
-
   import Record, only: [defrecordp: 2]
 
   defstruct [:topology_pid, :session, :doc, :cmd, :on_resume_token, :opts]
@@ -71,7 +69,6 @@ defmodule Mongo.ChangeStream do
 
     def aggregate(topology_pid, cmd, fun, opts) do
 
-      Logger.info("start_implicit_session ")
       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
 
@@ -87,13 +84,11 @@ defmodule Mongo.ChangeStream do
 
     def aggregate(topology_pid, session, doc, cmd, fun) do
 
-      Logger.info("aggregate wire version: #{inspect Mongo.wire_version(topology_pid)}")
       with %{"operationTime" => op_time,
              "cursor" => %{
                "id" => cursor_id,
                "ns" => coll,
-               "firstBatch" => docs} = response} <- doc,
-           {:ok, wire_version} <- Mongo.wire_version(topology_pid) do
+               "firstBatch" => docs} = response} <- doc do
 
         [%{"$changeStream" => stream_opts} | _pipeline] = Keyword.get(cmd, :pipeline) # extract the change stream options
 
@@ -105,7 +100,7 @@ defmodule Mongo.ChangeStream do
         # The initial aggregate response did not include a postBatchResumeToken.
 
         has_values = stream_opts["startAtOperationTime"] || stream_opts["startAfter"] || stream_opts["resumeAfter"]
-        op_time    = update_operation_time(op_time, has_values, docs, response["postBatchResumeToken"], wire_version)
+        op_time    = update_operation_time(op_time, has_values, docs, response["postBatchResumeToken"], Session.wire_version(session))
 
         # When the ChangeStream is started:
         # If startAfter is set, cache it.
@@ -132,7 +127,6 @@ defmodule Mongo.ChangeStream do
           change_stream(resume_token: resume_token, op_time: op_time, cmd: aggregate_cmd,
             on_resume_token: fun) = change_stream, opts) do
 
-      Logger.info("Get more")
       get_more = [
                    getMore: %BSON.LongNumber{value: cursor_id},
                    collection: coll,
@@ -159,26 +153,20 @@ defmodule Mongo.ChangeStream do
         {:error, %Mongo.Error{resumable: false} = not_resumable} -> {:error, not_resumable}
         {:error, _error} ->
 
-          Logger.info("Resuming....: #{inspect Mongo.wire_version(topology_pid) }")
-          with {:ok, wire_version} <- Mongo.wire_version(topology_pid) do
-
-            [%{"$changeStream" => stream_opts} | pipeline] = Keyword.get(aggregate_cmd, :pipeline) # extract the change stream options
+          [%{"$changeStream" => stream_opts} | pipeline] = Keyword.get(aggregate_cmd, :pipeline) # extract the change stream options
 
-            stream_opts   = update_stream_options(stream_opts, resume_token, op_time, wire_version)
-            aggregate_cmd = Keyword.update!(aggregate_cmd, :pipeline, fn _ -> [%{"$changeStream" => stream_opts} | pipeline] end)
+          stream_opts   = update_stream_options(stream_opts, resume_token, op_time, Session.wire_version(session))
+          aggregate_cmd = Keyword.update!(aggregate_cmd, :pipeline, fn _ -> [%{"$changeStream" => stream_opts} | pipeline] end)
 
-            # kill the cursor
-            kill_cursors(session, coll, [cursor_id], opts)
+          # kill the cursor
+          kill_cursors(session, coll, [cursor_id], opts)
 
-            # Start aggregation again...
-            Logger.info("Calling aggregate again")
-            with {:ok, state} <- aggregate(topology_pid, aggregate_cmd, fun, opts) do
-              {:resume, state}
-            end
+          # Start aggregation again...
+          with {:ok, state} <- aggregate(topology_pid, aggregate_cmd, fun, opts) do
+            {:resume, state}
           end
         reason ->
-          Logger.info("Error: #{inspect reason}")
-          {:error, nil}
+          {:error, reason}
       end
     end
 
@@ -260,7 +248,6 @@ defmodule Mongo.ChangeStream do
     """
     def kill_cursors(session, coll, cursor_ids, opts) do
 
-      ## todo Logger.info("Kill-Cursor")
       cmd = [
               killCursors: coll,
               cursors: cursor_ids |> Enum.map(fn id -> %BSON.LongNumber{value: id} end)

lib/mongo/session.ex

@@ -364,6 +364,14 @@ defmodule Mongo.Session do
 
   end
 
+  @doc """
+  Return the wire_version used in the session.
+  """
+  @spec connection(Session.t) :: pid
+  def wire_version(pid) do
+    call(pid, :wire_version)
+  end
+
   @doc """
   Return the connection used in the session.
   """
@@ -532,6 +540,9 @@ defmodule Mongo.Session do
   def handle_call_event(:abort_transaction, _state, _data) do
     {:keep_state_and_data, :ok}
   end
+  def handle_call_event(:wire_version, _state,  %{wire_version: wire_version}) do
+    {:keep_state_and_data, wire_version}
+  end
   def handle_call_event(:connection, _state,  %{conn: conn}) do
     {:keep_state_and_data, conn}
   end

lib/mongo/topology.ex

@@ -1,8 +1,6 @@
 defmodule Mongo.Topology do
   @moduledoc false
 
-  require Logger
-
   use GenServer
 
   alias Mongo.Events.ServerDescriptionChangedEvent
@@ -73,7 +71,6 @@ defmodule Mongo.Topology do
   end
 
   def mark_server_unknown(pid, address) do
-    ## todo Logger.info("mark_server_unknown #{inspect address} ")
     server_description = ServerDescription.from_is_master_error(address, "not writable primary or recovering")
     update_server_description(pid, server_description)
   end
@@ -226,7 +223,6 @@ defmodule Mongo.Topology do
   end
 
   def handle_info({:new_connection, waiting_pids}, state) do
-    ## todo Logger.info("Notify waitings pid with new_connection #{inspect waiting_pids}")
     Enum.each(waiting_pids, fn from -> GenServer.reply(from, :new_connection) end)
     {:noreply, state}
   end
@@ -255,20 +251,16 @@ defmodule Mongo.Topology do
       :empty ->
         case current == @min_heartbeat_frequency_ms do
           true ->
-            ## todo Logger.info("Topology: no primary found, searching")
             state
           false ->
-            ## todo Logger.info("Topology: no primary found, start searching")
             Enum.each(monitors, fn {_address, pid} -> Monitor.set_heartbeat_frequency_ms(pid, @min_heartbeat_frequency_ms) end)
             put_in(state[:topology][:heartbeat_frequency_ms], @min_heartbeat_frequency_ms)
         end
-      host ->
+      _host ->
         case current == @max_heartbeat_frequency_ms do
           true ->
-            ## todo Logger.info("Topology: primary exist")
             state
           false ->
-            ## todo Logger.info("Topology: primary found #{inspect host}, stop searching")
 
             ## filter own pid
             Enum.each(monitors, fn {_address, pid} -> Monitor.set_heartbeat_frequency_ms(pid, @max_heartbeat_frequency_ms) end)
@@ -351,7 +343,7 @@ defmodule Mongo.Topology do
     case TopologyDescription.select_servers(topology, :write, []) do
       :empty ->
         Mongo.Events.notify(%ServerSelectionEmptyEvent{action: :limits, cmd_type: :write, topology: topology})
-        {:reply, nil, state}
+        {:reply, {:error, :empty}, state}
       {:ok, {address, _opts}} ->
         with {:ok, limits} <- get_limits(address, topology) do
           {:reply, {:ok, limits}, state}
@@ -362,11 +354,10 @@ defmodule Mongo.Topology do
   end
 
   def handle_call(:wire_version, _from, %{:topology => topology} = state) do
-    case TopologyDescription.select_servers(topology, :read, []) do
+    case TopologyDescription.select_servers(topology, :write, []) do
       :empty ->
         Mongo.Events.notify(%ServerSelectionEmptyEvent{action: :wire_version, cmd_type: :read, topology: topology})
-        ## todo fix me
-        {:reply, {:ok, 9}, state}
+        {:reply, {:error, :empty}, state}
 
       {:ok, {address, _opts}} ->
         {:reply, {:ok, wire_version(address, topology)}, state}