ulfnet_amqpx

defmodule AMQPX.ConnectionPool do
  use GenServer
  require Logger

  @moduledoc """
  A pool of connections for shared use.

  AMQPX encourages using multiple channels per TCP connection instead of multiple connections, wherever possible.
  `AMQPX.ConnectionPool` stores open connections that other modules can retrieve from it.

  # Configuration

  ```elixir
  config :ulfnet_amqpx, connections: %{name => url}
  ```
  """

  defstruct connections: %{},
            config: []

  @doc false
  def child_spec(_) do
    %{
      id: __MODULE__,
      start: {__MODULE__, :start_link, []}
    }
  end

  @doc false
  def start_link(),
    do: GenServer.start(__MODULE__, nil, name: __MODULE__)

  @doc "Fetch a connection by its name."
  @spec get(name :: atom(), timeout :: integer() | :infinity) ::
          {:ok, AMQP.Connection.t()} | {:error, reason :: any()}
  def get(name, timeout \\ :infinity),
    do: GenServer.call(__MODULE__, {:get, name}, timeout)

  @doc "Register a connection config outside of `sys.config`."
  @spec register(name :: atom(), url :: any()) :: any()
  def register(name, url),
    do: GenServer.call(__MODULE__, {:register, name, url})

  @doc false
  def update_config,
    do: GenServer.call(__MODULE__, :update_config)

  @impl GenServer
  def init(_) do
    config = Application.get_env(:ulfnet_amqpx, :connections, [])
    state = %__MODULE__{config: config}
    {:ok, state}
  end

  @impl GenServer
  def handle_call({:get, id}, _, state = %__MODULE__{connections: connections, config: config}) do
    case connections do
      %{^id => conn} ->
        {:reply, {:ok, conn}, state}

      _ ->
        case try_connect(id, config) do
          resp = {:ok, conn} ->
            setup_monitor(id, conn)
            connections = Map.put(connections, id, conn)
            {:reply, resp, %__MODULE__{state | connections: connections}}

          err ->
            {:reply, err, state}
        end
    end
  end

  def handle_call(:update_config, _, state = %__MODULE__{config: config}) do
    new_config = Application.get_env(:ulfnet_amqpx, :connections, [])
    config = Map.merge(config, new_config)
    state = %__MODULE__{state | config: config}
    {:reply, config, state}
  end

  def handle_call({:register, name, url}, _, state = %__MODULE__{config: config}) do
    state = %__MODULE__{state | config: Map.put(config, name, url)}
    {:reply, :ok, state}
  end

  @impl GenServer
  def handle_info({:connection_down, id, _reason}, state = %__MODULE__{connections: connections}) do
    Logger.error("lost connection #{id}")
    connections = Map.delete(connections, id)
    {:noreply, %__MODULE__{state | connections: connections}}
  end

  defp try_connect(id, config) do
    case config do
      %{^id => url} ->
        AMQP.Connection.open(url)

      _ ->
        {:error, :not_configured}
    end
  end

  defp setup_monitor(id, conn) do
    ref = make_ref()
    pool = self()

    spawn(fn ->
      pid = conn.pid
      Process.monitor(pid)
      send(pool, {:monitor_setup, ref})

      receive do
        {:DOWN, _, :process, ^pid, reason} ->
          send(pool, {:connection_down, id, reason})
      end
    end)

    receive do
      {:monitor_setup, ^ref} -> nil
    end
  end
end