defmodule Spear.Connection do @moduledoc """ A GenServer which brokers a connection to an EventStore ## Configuration * `:name` - the name of the GenServer. See `t:GenServer.name/0` for more information. When not provided, the spawned process is not aliased to a name and is only addressable through its PID. * `:connection_string` - (**required**) the connection string to parse containing all connection information ## Examples iex> {:ok, conn} = Spear.Connection.start_link(connection_string: "esdb://localhost:2113") iex> Spear.stream!(conn, "es_supported_clients") |> Enum.take(3) [%Spear.Event{}, %Spear.Event{}, %Spear.Event{}] """ # see the very similar original implementation of this from the Mint # documentation: # https://github.com/elixir-mint/mint/blob/796b8db097d69ede7163acba223ab2045c2773a4/pages/Architecture.md use GenServer alias Spear.Connection.Request defstruct [:conn, requests: %{}] @typedoc """ A connection process A connection process (either referred to as `conn` or `connection` in the documentation) may either be a PID or a name such as a module or otherwise atom. ## Examples iex> {:ok, conn} = Spear.Connection.start_link(connection_string: "esdb://localhost:2113") {:ok, #PID<0.225.0>} iex> Spear.read_stream(conn, "es_supported_clients", max_count: 1) {:ok, #Stream<[ enum: #Function<62.80860365/2 in Stream.unfold/2>, funs: [#Function<48.80860365/1 in Stream.map/2>] ]>} """ @typedoc since: "0.1.0" @type t :: pid() | GenServer.name() @post "POST" @doc false def child_spec(init_arg) do default = %{ id: __MODULE__, start: {__MODULE__, :start_link, [init_arg]} } Supervisor.child_spec(default, []) end @doc """ Starts a connection process This function can be called directly in order to link it to the current process, but the more common workflow is to start a `Spear.Connection` GenServer as a part of a supervision tree. ## Examples E.g. in an application's supervision tree defined in `lib/my_app/application.ex`: children = [ {Spear.Connection, connection_string: "esdb://localhost:2113"} ] Supervisor.start_link(children, strategy: :one_for_one) """ @typedoc since: "0.1.0" @spec start_link(opts :: Keyword.t()) :: {:ok, t()} | GenServer.on_start() def start_link(opts) do name = Keyword.take(opts, [:name]) rest = Keyword.delete(opts, :name) GenServer.start_link(__MODULE__, rest, name) end @impl GenServer def init(config) do uri = config |> Keyword.fetch!(:connection_string) |> URI.parse() |> set_esdb_scheme() # YARD determine scheme from query params # YARD boot this to a handle_continue/2 or handle_cast/2? case Mint.HTTP.connect(uri.scheme, uri.host, uri.port, protocols: [:http2], mode: :active ) do {:ok, conn} -> {:ok, %__MODULE__{conn: conn}} {:error, reason} -> {:stop, reason} end end @impl GenServer def handle_call({:cancel, request_ref}, _from, state) when is_reference(request_ref) do with true <- Map.has_key?(state.requests, request_ref), {:ok, conn} <- Mint.HTTP2.cancel_request(state.conn, request_ref) do {:reply, :ok, put_in(state.conn, conn)} else # coveralls-ignore-start false -> # idempotent success when the request_ref is not active {:reply, :ok, state} {:error, conn, reason} -> {:reply, {:error, reason}, put_in(state.conn, conn)} # coveralls-ignore-stop end end def handle_call({type, request}, from, state) do case request_and_stream_body(state, request, from, type) do {:ok, state} -> {:noreply, state} # coveralls-ignore-start {:error, state, reason} -> {:reply, {:error, reason}, state} # coveralls-ignore-stop end end @impl GenServer def handle_info(message, %{conn: conn} = state) do case Mint.HTTP2.stream(conn, message) do :unknown -> {:noreply, state} {:ok, conn, responses} -> state = put_in(state.conn, conn) {:noreply, handle_responses(state, responses)} {:error, conn, _reason, responses} -> # coveralls-ignore-start # YARD error handling state = put_in(state.conn, conn) {:noreply, handle_responses(state, responses)} # coveralls-ignore-stop end end @spec handle_responses(%__MODULE__{}, list()) :: %__MODULE__{} defp handle_responses(state, responses) do responses |> Enum.reduce(state, &process_response/2) |> Request.continue_requests() end defp process_response({:status, request_ref, status}, state) do put_in(state.requests[request_ref].response.status, status) end defp process_response({:headers, request_ref, new_headers}, state) do update_in( state.requests[request_ref].response.headers, fn headers -> headers ++ new_headers end ) end defp process_response({:data, request_ref, new_data}, state) do update_in( state.requests[request_ref], &Request.handle_data(&1, new_data) ) end defp process_response({:done, request_ref}, state) do {%{response: response, from: from}, state} = pop_in(state.requests[request_ref]) GenServer.reply(from, {:ok, response}) state end # coveralls-ignore-start defp process_response(_unknown, state), do: state # coveralls-ignore-stop defp request_and_stream_body(state, request, from, request_type) do with {:ok, conn, request_ref} <- Mint.HTTP2.request(state.conn, @post, request.path, request.headers, :stream), request = Request.new(request, request_ref, from, request_type), state = put_in(state.conn, conn), state = put_in(state.requests[request_ref], request), {:ok, state} <- Request.emit_messages(state, request) do {:ok, state} else # coveralls-ignore-start {:error, %__MODULE__{} = state, reason} -> {:error, state, reason} {:error, conn, reason} -> {:error, put_in(state.conn, conn), reason} # coveralls-ignore-stop end end defp set_esdb_scheme(%URI{scheme: "esdb"} = uri), do: %URI{uri | scheme: :http} defp set_esdb_scheme(%URI{scheme: "http"} = uri), do: %URI{uri | scheme: :http} end