defmodule Mongo.StreamingHelloMonitor do
@moduledoc """
See https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-monitoring.rst#streaming-protocol
The streaming protocol is used to monitor MongoDB 4.4+ servers and optimally reduces the time it takes for a client to discover server state changes.
Multi-threaded or asynchronous drivers MUST use the streaming protocol when connected to a server that supports the awaitable hello or legacy hello commands.
This protocol requires an extra thread and an extra socket for each monitor to perform RTT calculations.
This module implements the streaming protocol. The GenServer is started and maintained by the Monitor process. The streaming hello monitor uses the
more to come flag while updating the current server description.
"""
require Logger
use GenServer
import Bitwise
alias Mongo.Events.ServerHeartbeatFailedEvent
alias Mongo.Events.ServerHeartbeatStartedEvent
alias Mongo.Events.ServerHeartbeatSucceededEvent
alias Mongo.ServerDescription
alias Mongo.StreamingHelloMonitor
alias Mongo.Topology
@more_to_come_mask 0x2
def start_link(args) do
GenServer.start_link(StreamingHelloMonitor, args)
end
@doc """
Initialize the monitor process
"""
def init([topology_pid, address, opts]) do
heartbeat_frequency_ms = 10_000
opts =
opts
|> Keyword.drop([:after_connect])
|> Keyword.put(:after_connect, {StreamingHelloMonitor, :connected, [self()]})
|> Keyword.put(:connection_type, :stream_monitor)
## debug info("Starting stream hello monitor with options #{inspect(opts, pretty: true)}")
with {:ok, pid} <- DBConnection.start_link(Mongo.MongoDBConnection, opts) do
{:ok,
%{
## our connection pid to the mongodb server
connection_pid: pid,
## the topology_pid to which we report
topology_pid: topology_pid,
## the address of the server, needed to make updates
address: address,
## current heartbeat_frequency_ms
heartbeat_frequency_ms: heartbeat_frequency_ms,
max_await_time_ms: heartbeat_frequency_ms,
more_to_come: false,
topology_version: nil,
## options
opts: opts
}}
end
end
@doc """
In this case we stop the DBConnection.
"""
def terminate(reason, %{connection_pid: connection_pid}) do
## debug info("Terminating streaming hello monitor for reason #{inspect reason}")
GenServer.stop(connection_pid, reason)
end
@doc """
Report the connection event, so the topology process can now create the connection pool.
"""
def connected(_connection, me) do
GenServer.cast(me, :update)
end
@doc """
Time to update the topology. Calling `hello` and updating the server description
"""
def handle_cast(:update, state) do
handle_info(:update, state)
end
def handle_info({:EXIT, _pid, reason}, state) do
## debug Logger.warn("Stopped with reason #{inspect reason}")
{:stop, reason, state}
end
def handle_info(:update, state) do
state = update_server_description(state)
case state.more_to_come do
true ->
send(self(), :update)
false ->
:noop
end
{:noreply, state}
end
##
# Get a new server description from the server and send it to the Topology process.
#
defp update_server_description(%{topology_pid: topology_pid} = state) do
with {topology_version, flags, server_description} <- get_server_description(state) do
## debug info("Updating server description")
Topology.update_server_description(topology_pid, server_description)
state = %{state | topology_version: topology_version}
case flags &&& @more_to_come_mask do
@more_to_come_mask ->
## debug info("More to come...")
%{state | more_to_come: true}
_other ->
Process.send_after(self(), :update, state.heartbeat_frequency_ms)
%{state | more_to_come: false}
end
end
end
# see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#network-error-when-calling-ismaster
##
# Calls hello command and parses the result to update the server description
#
defp get_server_description(%{connection_pid: conn_pid, address: address, topology_version: topology_version, max_await_time_ms: max_await_time_ms, opts: opts} = state) do
Mongo.Events.notify(%ServerHeartbeatStartedEvent{connection_pid: conn_pid})
opts = Keyword.merge(opts, timeout: max_await_time_ms * 2)
{duration, result} =
case state do
%{more_to_come: true} ->
:timer.tc(fn -> Mongo.exec_more_to_come(conn_pid, opts) end)
_other ->
opts = Keyword.put(opts, :max_await_time_ms, max_await_time_ms)
:timer.tc(fn -> hello_command(conn_pid, topology_version, opts) end)
end
case result do
{:ok, {flags, hello_doc}} ->
server_description =
hello_doc
|> ServerDescription.parse_hello_response()
|> Map.put(:address, address)
|> Map.put(:last_update_time, DateTime.utc_now())
|> Map.put(:error, nil)
notify_success(duration, hello_doc, conn_pid)
{hello_doc["topologyVersion"], flags, server_description}
{:error, error} ->
notify_error(duration, error, conn_pid)
server_description =
ServerDescription.new()
|> Map.put(:address, address)
|> Map.put(:error, error)
{nil, 0x0, server_description}
end
end
defp hello_command(conn_pid, %{"counter" => counter, "processId" => process_id}, opts) do
max_await_time_ms = Keyword.get(opts, :max_await_time_ms, 10_000)
opts =
opts
|> Keyword.merge(flags: [:exhaust_allowed])
|> Keyword.merge(timeout: max_await_time_ms * 2)
cmd = [
maxAwaitTimeMS: max_await_time_ms,
topologyVersion: %{
counter: %BSON.LongNumber{value: counter},
processId: process_id
}
]
Mongo.exec_hello(conn_pid, cmd, opts)
end
defp hello_command(conn_pid, _topology_version, opts) do
Mongo.exec_hello(conn_pid, opts)
end
defp notify_error(duration, error, conn_pid) do
Mongo.Events.notify(%ServerHeartbeatFailedEvent{duration: duration, failure: error, connection_pid: conn_pid})
end
defp notify_success(duration, reply, conn_pid) do
Mongo.Events.notify(%ServerHeartbeatSucceededEvent{duration: duration, reply: reply, connection_pid: conn_pid})
end
def info(message) do
Logger.info(IO.ANSI.format([:blue, :bright, message]))
end
end
