add support for the new hello handshake, resolves #140 (#143)

Author: zookzook

CHANGELOG.md

@@ -5,9 +5,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## 0.9.2 (2022-09-07)
+## 0.9.2 (2022-09-xx)
 * Bugfix
   * fix a crash in the streaming hello monitor, if the server sends more than one response at once 
+  * add support for the new hello handshake
 
 ## 0.9.1 (2022-05-27)
 * Bugfix

lib/mongo.ex

@@ -1523,6 +1523,18 @@ defmodule Mongo do
     end
   end
 
+  def exec_hello(conn, opts) do
+    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:exec_hello, []}}, [], defaults(opts)) do
+      check_for_error(response, [hello: 1], opts)
+    end
+  end
+
+  def exec_hello(conn, cmd, opts) do
+    with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: {:exec_hello, cmd}}, [], defaults(opts)) do
+      check_for_error(response, cmd, opts)
+    end
+  end
+
   def exec_more_to_come(conn, opts) do
     with {:ok, _cmd, response} <- DBConnection.execute(conn, %Query{action: :command}, [:more_to_come], defaults(opts)) do
       check_for_error(response, [:more_to_come], opts)
@@ -1745,7 +1757,7 @@ defmodule Mongo do
   end
 
   defp do_log(cmd, duration, opts) do
-    case Keyword.has_key?(cmd, :isMaster) || Keyword.has_key?(cmd, :more_to_come) do
+    case Keyword.has_key?(cmd, :isMaster) || Keyword.has_key?(cmd, :more_to_come) || Keyword.has_key?(cmd, :hello) do
       true ->
         :ok
 

lib/mongo/auth/scram.ex

@@ -127,7 +127,8 @@ defmodule Mongo.Auth.SCRAM do
   # It calls isMaster with saslSupportedMechs option to ask for the selected user which mechanism is supported
   #
   defp select_digest(database, username, state) do
-    with {:ok, _flags, doc} <- Utils.command(-2, [isMaster: 1, saslSupportedMechs: database <> "." <> username], state) do
+    ### todo
+    with {:ok, _flags, doc} <- Utils.command(-2, [hello: 1, saslSupportedMechs: database <> "." <> username], state) do
       select_digest(doc)
     end
   end

lib/mongo/mongo_db_connection.ex

@@ -12,11 +12,12 @@ defmodule Mongo.MongoDBConnection do
   alias Mongo.Events.CommandStartedEvent
   alias Mongo.MongoDBConnection.Utils
   alias Mongo.Events.MoreToComeEvent
+  alias Mongo.StableVersion
 
   @timeout 5_000
   @find_one_flags ~w(slave_ok exhaust partial)a
   @write_concern ~w(w j wtimeout)a
-  @insecure_cmds [:authenticate, :saslStart, :saslContinue, :getnonce, :createUser, :updateUser, :copydbgetnonce, :copydbsaslstart, :copydb, :isMaster, :ismaster]
+  @insecure_cmds [:authenticate, :saslStart, :saslContinue, :getnonce, :createUser, :updateUser, :copydbgetnonce, :copydbsaslstart, :copydb, :isMaster, :ismaster, :hello]
 
   @impl true
   def connect(opts) do
@@ -34,12 +35,74 @@ defmodule Mongo.MongoDBConnection do
       auth_mechanism: opts[:auth_mechanism] || nil,
       connection_type: Keyword.fetch!(opts, :connection_type),
       topology_pid: Keyword.fetch!(opts, :topology_pid),
+      stable_api: Keyword.get(opts, :stable_api),
+      use_op_msg: Keyword.get(opts, :stable_api) != nil,
+      hello_ok: Keyword.get(opts, :stable_api) != nil,
       ssl: opts[:ssl] || opts[:tls] || false
     }
 
     connect(opts, state)
   end
 
+  @impl true
+  def disconnect(_error, %{connection: {mod, socket}, connection_type: type, topology_pid: pid, host: host}) do
+    GenServer.cast(pid, {:disconnect, type, host})
+    mod.close(socket)
+    :ok
+  end
+
+  @impl true
+  def checkout(state), do: {:ok, state}
+  @impl true
+  def handle_begin(_opts, state), do: {:ok, nil, state}
+  @impl true
+  def handle_close(_query, _opts, state), do: {:ok, nil, state}
+  @impl true
+  def handle_commit(_opts, state), do: {:ok, nil, state}
+  @impl true
+  def handle_deallocate(_query, _cursor, _opts, state), do: {:ok, nil, state}
+  @impl true
+  def handle_declare(query, _params, _opts, state), do: {:ok, query, nil, state}
+  @impl true
+  def handle_fetch(_query, _cursor, _opts, state), do: {:halt, nil, state}
+  @impl true
+  def handle_prepare(query, _opts, state), do: {:ok, query, state}
+  @impl true
+  def handle_rollback(_opts, state), do: {:ok, nil, state}
+  @impl true
+  def handle_status(_opts, state), do: {:idle, state}
+
+  @impl true
+  def ping(%{connection_type: :client} = state) do
+    cmd = [ping: 1]
+
+    case Utils.command(-1, cmd, state) do
+      {:ok, _flags, %{"ok" => ok}} when ok == 1 ->
+        {:ok, state}
+
+      {:ok, _flags, %{"ok" => ok, "errmsg" => msg, "code" => code}} when ok == 0 ->
+        err = Mongo.Error.exception(message: msg, code: code)
+        {:disconnect, err, state}
+
+      {:disconnect, _, _} = error ->
+        error
+    end
+  end
+
+  @impl true
+  def ping(state) do
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_execute(%Mongo.Query{action: action} = query, params, opts, original_state) do
+    tmp_state = %{original_state | database: Keyword.get(opts, :database, original_state.database)}
+
+    with {:ok, reply, tmp_state} <- execute_action(action, params, opts, tmp_state) do
+      {:ok, query, reply, Map.put(tmp_state, :database, original_state.database)}
+    end
+  end
+
   defp connect(opts, state) do
     result =
       with {:ok, state} <- tcp_connect(opts, state),
@@ -123,12 +186,12 @@ defmodule Mongo.MongoDBConnection do
     end
   end
 
-  defp post_hello_command(state, client) do
-    cmd = [ismaster: 1, client: client]
+  defp hand_shake(opts, state) do
+    cmd = handshake_command(state, client(opts[:appname] || "elixir-driver"))
 
     case Utils.command(-1, cmd, state) do
-      {:ok, _flags, %{"ok" => ok, "maxWireVersion" => version}} when ok == 1 ->
-        {:ok, %{state | wire_version: version}}
+      {:ok, _flags, %{"ok" => ok, "maxWireVersion" => version} = response} when ok == 1 ->
+        {:ok, %{state | wire_version: version, use_op_msg: version >= 6, hello_ok: Map.get(response, "helloOk", false)}}
 
       {:ok, _flags, %{"ok" => ok}} when ok == 1 ->
         {:ok, %{state | wire_version: 0}}
@@ -142,11 +205,7 @@ defmodule Mongo.MongoDBConnection do
     end
   end
 
-  defp hand_shake(opts, state) do
-    post_hello_command(state, driver(opts[:appname] || "My killer app"))
-  end
-
-  defp driver(appname) do
+  defp client(app_name) do
     driver_version =
       case :application.get_key(:mongodb_driver, :vsn) do
         {:ok, version} -> to_string(version)
@@ -167,7 +226,7 @@ defmodule Mongo.MongoDBConnection do
 
     %{
       client: %{
-        application: %{name: appname}
+        application: %{name: app_name}
       },
       driver: %{
         name: "mongodb_driver",
@@ -195,76 +254,50 @@ defmodule Mongo.MongoDBConnection do
   defp pretty_name("apple"), do: "Mac OS X"
   defp pretty_name(name), do: name
 
-  @impl true
-  def disconnect(_error, %{connection: {mod, socket}, connection_type: type, topology_pid: pid, host: host}) do
-    GenServer.cast(pid, {:disconnect, type, host})
-    mod.close(socket)
-    :ok
+  defp provide_cmd_data([{command_name, _} | _] = cmd) do
+    case Enum.member?(@insecure_cmds, command_name) do
+      true -> {command_name, %{}}
+      false -> {command_name, cmd}
+    end
   end
 
-  @impl true
-  def checkout(state), do: {:ok, state}
-  @impl true
-  def handle_begin(_opts, state), do: {:ok, nil, state}
-  @impl true
-  def handle_close(_query, _opts, state), do: {:ok, nil, state}
-  @impl true
-  def handle_commit(_opts, state), do: {:ok, nil, state}
-  @impl true
-  def handle_deallocate(_query, _cursor, _opts, state), do: {:ok, nil, state}
-  @impl true
-  def handle_declare(query, _params, _opts, state), do: {:ok, query, nil, state}
-  @impl true
-  def handle_fetch(_query, _cursor, _opts, state), do: {:halt, nil, state}
-  @impl true
-  def handle_prepare(query, _opts, state), do: {:ok, query, state}
-  @impl true
-  def handle_rollback(_opts, state), do: {:ok, nil, state}
-  @impl true
-  def handle_status(_opts, state), do: {:idle, state}
-
-  @impl true
-  def ping(%{connection_type: :client} = state) do
-    cmd = [ping: 1]
-
-    case Utils.command(-1, cmd, state) do
-      {:ok, _flags, %{"ok" => ok}} when ok == 1 ->
-        {:ok, state}
+  ##
+  # Executes a hello or the legacy hello command
+  ##
+  defp execute_action({:exec_hello, cmd}, _params, opts, %{use_op_msg: true} = state) do
+    db = opts[:database] || state.database
+    timeout = Keyword.get(opts, :timeout, state.timeout)
+    flags = Keyword.get(opts, :flags, 0x0)
 
-      {:ok, _flags, %{"ok" => ok, "errmsg" => msg, "code" => code}} when ok == 0 ->
-        err = Mongo.Error.exception(message: msg, code: code)
-        {:disconnect, err, state}
+    cmd = hello_command(cmd, state) ++ ["$db": db]
 
-      {:disconnect, _, _} = error ->
-        error
-    end
-  end
+    event = %CommandStartedEvent{
+      command: :hello,
+      command_name: :hello,
+      database_name: db,
+      request_id: state.request_id,
+      operation_id: opts[:operation_id],
+      connection_id: self()
+    }
 
-  @impl true
-  def ping(state) do
-    {:ok, state}
-  end
+    Events.notify(event, :commands)
 
-  @impl true
-  def handle_execute(%Mongo.Query{action: action} = query, params, opts, original_state) do
-    tmp_state = %{original_state | database: Keyword.get(opts, :database, original_state.database)}
+    op = op_msg(flags: flags, sections: [section(payload_type: 0, payload: payload(doc: cmd))])
 
-    with {:ok, reply, tmp_state} <- execute_action(action, params, opts, tmp_state) do
-      {:ok, query, reply, Map.put(tmp_state, :database, original_state.database)}
-    end
-  end
+    case :timer.tc(fn -> Utils.post_request(op, state.request_id, %{state | timeout: timeout}) end) do
+      {duration, {:ok, flags, doc}} ->
+        state = %{state | request_id: state.request_id + 1}
+        {:ok, {doc, event, flags, duration}, state}
 
-  defp provide_cmd_data([{command_name, _} | _] = cmd) do
-    case Enum.member?(@insecure_cmds, command_name) do
-      true -> {command_name, %{}}
-      false -> {command_name, cmd}
+      {_duration, error} ->
+        error
     end
   end
 
   ##
   # Executes a more to come command
   ##
-  defp execute_action(:command, [:more_to_come], opts, %{wire_version: version} = state) when version >= 6 do
+  defp execute_action(:command, [:more_to_come], opts, %{use_op_msg: true} = state) do
     event = %MoreToComeEvent{command: :more_to_come, command_name: opts[:command_name] || :more_to_come}
 
     timeout = Keyword.get(opts, :timeout, state.timeout)
@@ -280,7 +313,7 @@ defmodule Mongo.MongoDBConnection do
     end
   end
 
-  defp execute_action(:command, [cmd], opts, %{wire_version: version} = state) when version >= 6 do
+  defp execute_action(:command, [cmd], opts, %{use_op_msg: true} = state) do
     {command_name, data} = provide_cmd_data(cmd)
     db = opts[:database] || state.database
     cmd = cmd ++ ["$db": db]
@@ -371,4 +404,24 @@ defmodule Mongo.MongoDBConnection do
       {_flag, false}, acc -> acc
     end)
   end
+
+  defp handshake_command(%{stable_api: nil}, client) do
+    [ismaster: 1, helloOk: 1, client: client]
+  end
+
+  defp handshake_command(%{stable_api: stable_api}, client) do
+    StableVersion.merge_stable_api([hello: 1, client: client], stable_api)
+  end
+
+  defp hello_command(cmd, %{hello_ok: false}) do
+    cmd
+    |> Keyword.put(:ismaster, 1)
+    |> Keyword.put(:helloOk, 1)
+  end
+
+  defp hello_command(cmd, %{hello_ok: true, stable_api: stable_api}) do
+    cmd
+    |> Keyword.put(:hello, 1)
+    |> StableVersion.merge_stable_api(stable_api)
+  end
 end

lib/mongo/monitor.ex

@@ -221,7 +221,7 @@ defmodule Mongo.Monitor do
   # Streaming mode: calls hello command and updated the round trip time for the command.
   ##
   defp get_server_description(%{connection_pid: conn_pid, round_trip_time: last_rtt, mode: :streaming_mode, opts: opts}) do
-    {rtt, response} = :timer.tc(fn -> Mongo.exec_command(conn_pid, [isMaster: 1], opts) end)
+    {rtt, response} = :timer.tc(fn -> Mongo.exec_hello(conn_pid, opts) end)
 
     case response do
       {:ok, {_flags, _hello_doc}} ->
@@ -236,7 +236,7 @@ defmodule Mongo.Monitor do
   # Polling mode: updating the server description and the round trip time together
   ##
   defp get_server_description(%{connection_pid: conn_pid, address: address, round_trip_time: last_rtt, opts: opts}) do
-    {rtt, response} = :timer.tc(fn -> Mongo.exec_command(conn_pid, [isMaster: 1], opts) end)
+    {rtt, response} = :timer.tc(fn -> Mongo.exec_hello(conn_pid, opts) end)
 
     case response do
       {:ok, {_flags, hello_doc}} ->

lib/mongo/query.ex

@@ -2,7 +2,7 @@ defmodule Mongo.Query do
   @moduledoc """
     This is the query implementation for the Query Protocol
 
-    Encoding and decoding does not take place at this point, but is directly performed
+  Encoding and decoding does not take place at this point, but is directly performed
     into the functions of Mongo.MongoDBConnection.Utils.
   """
   defstruct action: nil

lib/mongo/server_description.ex

@@ -151,6 +151,7 @@ defmodule Mongo.ServerDescription do
   defp determine_server_type(%{"setName" => set_name} = is_master_reply) when set_name != nil do
     case is_master_reply do
       %{"ismaster" => true} -> :rs_primary
+      %{"isWritablePrimary" => true} -> :rs_primary
       %{"secondary" => true} -> :rs_secondary
       %{"arbiterOnly" => true} -> :rs_arbiter
       _ -> :rs_other

lib/mongo/stable_version.ex

@@ -0,0 +1,22 @@
+defmodule Mongo.StableVersion do
+  @moduledoc false
+
+  defmodule ServerAPI do
+    @moduledoc false
+
+    defstruct version: "1",
+              strict: false,
+              deprecation_errors: false
+  end
+
+  def merge_stable_api(command, %{version: version, strict: strict, deprecation_errors: deprecation_errors}) do
+    command
+    |> Keyword.put(:apiVersion, version)
+    |> Keyword.put(:apiStrict, strict)
+    |> Keyword.put(:apiDeprecationErrors, deprecation_errors)
+  end
+
+  def merge_stable_api(command, _other) do
+    command
+  end
+end