rabbit_mq

defmodule RabbitMQ.Connection do
  @moduledoc """
  Convenience to open and maintain a `AMQP.Connection`.

  Should be started under a `Supervisor`.
  """

  use AMQP
  use GenServer

  require Logger

  @connection_opts ~w(heartbeat_interval_sec max_channels name reconnect_interval_ms)a
  @this_module __MODULE__

  ##############
  # Public API #
  ##############

  @spec start_link(keyword()) :: GenServer.on_start()
  def start_link(opts) do
    name = Keyword.fetch!(opts, :name)
    connection_opts = Keyword.take(opts, @connection_opts)

    GenServer.start_link(@this_module, connection_opts, name: name)
  end

  @doc """
  Retrieves the `AMQP.Connection` from a given `RabbitMQ.Connection` server.
  """
  @spec get(atom()) :: Connection.t()
  def get(name) when is_atom(name) do
    GenServer.call(name, :get)
  end

  ######################
  # Callback Functions #
  ######################

  @impl true
  def init(connection_opts) do
    # This is needed to invoke `terminate/2` when the parent process,
    # ideally a `Supervisor`, sends an exit signal.
    #
    # Read more @ https://hexdocs.pm/elixir/GenServer.html#c:terminate/2.
    Process.flag(:trap_exit, true)

    connect(connection_opts)
  end

  @impl true
  def handle_call(:get, _from, %Connection{} = connection), do: {:reply, connection, connection}

  @impl true
  def handle_info({:DOWN, _, :process, _pid, reason}, %Connection{} = connection) do
    Logger.warn("Connection to broker lost due to #{inspect(reason)}.")

    # Stop GenServer; will be restarted by Supervisor.
    {:stop, {:connection_lost, reason}, connection}
  end

  @doc """
  Invoked when the server is about to exit. It should do any cleanup required.
  See https://hexdocs.pm/elixir/GenServer.html#c:terminate/2 for more details.
  """
  @impl true
  def terminate(reason, %Connection{} = connection) do
    Logger.warn("Terminating Connection server: #{inspect(reason)}.")

    # Not sure this check is needed.
    if Process.alive?(connection.pid) do
      :ok = Connection.close(connection)
    end

    {:noreply, connection}
  end

  #####################
  # Private Functions #
  #####################

  defp connect(connection_opts) do
    name = connection_name(connection_opts)

    heartbeat_interval_sec = Keyword.get(connection_opts, :heartbeat_interval_sec, 30)
    max_channels = Keyword.get(connection_opts, :max_channels, 8)
    reconnect_interval_ms = Keyword.get(connection_opts, :reconnect_interval_ms, 2500)

    opts = [channel_max: max_channels, heartbeat: heartbeat_interval_sec]

    amqp_url()
    |> Connection.open(name, opts)
    |> case do
      {:ok, connection} ->
        Logger.debug("Connected to broker")

        # Monitor the connection process. Should the process go down,
        # we will try to exit cleanly and let the supervisor restart the process.
        _ref = Process.monitor(connection.pid)

        {:ok, connection}

      {:error, error} ->
        Logger.error("Failed to connect to broker due to #{inspect(error)}. Retrying...")
        :timer.sleep(reconnect_interval_ms)
        connect(connection_opts)
    end
  end

  defp amqp_url, do: Application.fetch_env!(:rabbit_mq, :amqp_url)

  defp connection_name(connection_opts) do
    {:ok, name} = Keyword.fetch(connection_opts, :name)
    {:ok, hostname} = :inet.gethostname()
    "#{hostname}/#{name}"
  end
end